{"record":{"id":"b104930337721a88","repo":"pinpoint-apm/pinpoint","slug":"not-found-b10493","errorCode":"NOT_FOUND","errorMessage":"Could not find echo sink: clusterKey = {}","messagePattern":"Could not find echo sink: clusterKey = (.+?)","errorType":"error_code","errorClass":"StatusException","httpStatus":null,"severity":"error","filePath":"realtime/realtime-collector/src/main/java/com/navercorp/pinpoint/realtime/collector/receiver/grpc/GrpcCommandService.java","lineNumber":306,"sourceCode":"        long sinkId = response.getCommonResponse().getResponseId();\n        final ActiveThreadLightDumpPublisher publisher = this.activeThreadLightDumpSinkRepo.get(sinkId);\n        emitMono(response, responseObserver, publisher);\n        this.activeThreadLightDumpSinkRepo.invalidate(sinkId);\n    }\n\n    @Override\n    public StreamObserver<PCmdActiveThreadCountRes> commandStreamActiveThreadCount(StreamObserver<Empty> responseObserver) {\n        logger.debug(\"commandStreamActiveThreadCount started\");\n\n        ServerCallStreamObserver<Empty> serverResponseObserver = (ServerCallStreamObserver<Empty>) responseObserver;\n        return new ActiveThreadCountResponseStreamObserver(serverResponseObserver, this.activeThreadCountSinkRepo);\n    }\n\n    private <T> void emitMono(T response, StreamObserver<Empty> responseObserver, Publisher<T> sink) {\n        if (sink == null) {\n            if (logger.isWarnEnabled()) {\n                Header header = ServerContext.getAgentInfo();\n                logger.warn(\"Could not find echo sink: clusterKey = {}\", getClusterKeyFromContext(header));\n            }\n            responseObserver.onError(new StatusException(Status.NOT_FOUND));\n            return;\n        }\n        sink.publish(response);\n        responseObserver.onNext(Empty.getDefaultInstance());\n        responseObserver.onCompleted();\n    }\n\n    private InetSocketAddress getRemoteAddressFromContext() {\n        TransportMetadata transportMetadata = ServerContext.getTransportMetadata();\n        return transportMetadata.getRemoteAddress();\n    }\n\n    private ClusterKey getClusterKeyFromContext(Header header) {\n        return new ClusterKey(header.getServiceName(), header.getApplicationName(), header.getAgentId(), header.getAgentStartTime());\n    }\n","sourceCodeStart":288,"sourceCodeEnd":324,"githubUrl":"https://github.com/pinpoint-apm/pinpoint/blob/744c3d3075e595656abb1ae331ad2c0e4c9eb996/realtime/realtime-collector/src/main/java/com/navercorp/pinpoint/realtime/collector/receiver/grpc/GrpcCommandService.java#L288-L324","documentation":"When the collector receives a command echo/activeThreadDump/activeThreadLightDump response from an agent, emitMono looks up the registered sink (Publisher) for that agent connection. If no sink is registered for the connection identified by the request's cluster key, it logs 'Could not find echo sink' and returns gRPC Status.NOT_FOUND to the caller. This means the collector has no active publisher session for that agent's command responses.","triggerScenarios":"A command response arrives from (or is emitted for) an agent whose sink was never created or was already removed from the SinkRepository — e.g. the agent's command stream disconnected and cleaned up sinks before the response was published, or the command was issued for an agent not currently connected to this collector instance.","commonSituations":"Web/collector topology mismatch: the command was sent through one collector node but the agent's stream is attached to another (no shared sink repo); agent reconnected between request and response; stale cluster metadata in the web UI issuing commands to a dead agent session.","solutions":["Retry the command after confirming the agent is currently connected to the same collector instance that handles it","Ensure all collector nodes share the sink registry or that command routing is sticky to the node holding the agent's stream","Refresh the agent list in the web/cluster metadata so commands are not issued for stale agent sessions","Check collector logs for an earlier disconnect of the agent's command stream (which removes the sinks)"],"exampleFix":"// caller-side guard before invoking commandEcho\nEchoPublisher sink = echoSinkRepo.getEchoPublisher(clusterKey);\nif (sink == null) {\n    // agent stream not connected to this collector; re-fetch or fail fast\n    throw new AgentNotConnectedException(clusterKey);\n}\nprofilerCommandService.commandEcho(responseObserver, echoRequest);","handlingStrategy":"fallback","validationCode":"// before issuing a command, verify the sink exists on this collector\nPublisher<?> sink = sinkRepository.getPublisher(clusterKey);\nif (sink == null) {\n    throw new AgentNotConnectedException(\"no sink registered for \" + clusterKey);\n}","typeGuard":null,"tryCatchPattern":"try {\n    profilerCommandService.commandEcho(responseObserver, echoRequest);\n} catch (StatusRuntimeException e) {\n    if (e.getStatus().getCode() == Status.Code.NOT_FOUND) {\n        log.warn(\"Agent session gone; refresh cluster metadata and retry\", e);\n        retryAfterRefresh(echoRequest);\n    }\n}","preventionTips":["Confirm the agent's command stream is connected to the collector node handling the command (sticky routing or shared sink repo)","Refresh cluster/agent metadata before issuing commands so dead sessions are not targeted","Handle agent reconnects: reissue commands after the new stream (and sinks) are registered","Monitor sink registration/removal logs to detect premature sink cleanup"],"tags":["grpc","pinpoint","sink-registry","session-expired"],"backgroundTag":"resource-not-found","analyzedSha":"744c3d3075e595656abb1ae331ad2c0e4c9eb996","analyzedAt":"2026-09-07T18:48:45.289Z","contentChangedAt":"2026-09-07T18:48:45.289Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}