{"record":{"id":"b25dd8b2f7ea7ee3","repo":"pinpoint-apm/pinpoint","slug":"failed-to-emit-complete-sink-not-found","errorCode":null,"errorMessage":"Failed to emit complete: sink not found","messagePattern":"Failed to emit complete: sink not found","errorType":"console","errorClass":null,"httpStatus":null,"severity":"error","filePath":"realtime/realtime-collector/src/main/java/com/navercorp/pinpoint/realtime/collector/receiver/grpc/ActiveThreadCountResponseStreamObserver.java","lineNumber":115,"sourceCode":"        this.serverCallStreamObserver.onCompleted();\n\n        if (this.publisher != null) {\n            this.publisher.error(t);\n        } else {\n            logger.warn(\"Failed to emit error: sink not found. the error may have occurred before the first message\");\n        }\n    }\n\n    @Override\n    public void onCompleted() {\n        logger.info(\"Completed stream: sinkId = {}\", this.sinkId);\n\n        this.serverCallStreamObserver.onCompleted();\n\n        if (this.publisher != null) {\n            this.publisher.complete();\n        } else {\n            logger.warn(\"Failed to emit complete: sink not found\");\n        }\n    }\n\n    private long extractSinkId(PCmdActiveThreadCountRes response) {\n        return response.getCommonStreamResponse().getResponseId();\n    }\n\n    private int extractSequence(PCmdActiveThreadCountRes response) {\n        return response.getCommonStreamResponse().getSequenceId();\n    }\n\n}","sourceCodeStart":97,"sourceCodeEnd":127,"githubUrl":"https://github.com/pinpoint-apm/pinpoint/blob/744c3d3075e595656abb1ae331ad2c0e4c9eb996/realtime/realtime-collector/src/main/java/com/navercorp/pinpoint/realtime/collector/receiver/grpc/ActiveThreadCountResponseStreamObserver.java#L97-L127","documentation":"Warn log in ActiveThreadCountResponseStreamObserver.onCompleted: the agent's stream finished before a publisher was ever bound (no sinkId/publisher established), so the completion signal has no reactive sink to complete and any waiting subscriber would hang.","triggerScenarios":"Thrown at realtime/realtime-collector/src/main/java/com/navercorp/pinpoint/realtime/collector/receiver/grpc/ActiveThreadCountResponseStreamObserver.java:115 when the library encounters an invalid state.","commonSituations":"See trigger scenarios.","solutions":["Ensure the first response carries a valid sinkId so initSink binds a publisher","Guard subscribers with a timeout so they fail rather than hang","Log the sinkId (if any) to correlate with the issuing request"],"exampleFix":null,"handlingStrategy":"fallback","validationCode":null,"typeGuard":null,"tryCatchPattern":null,"preventionTips":[],"tags":[],"backgroundTag":null,"analyzedSha":"744c3d3075e595656abb1ae331ad2c0e4c9eb996","analyzedAt":"2026-09-07T18:48:45.289Z","contentChangedAt":"2026-09-07T18:48:45.289Z","schemaVersion":2},"datasetVersion":"2026-09-17T15:17:12.973Z"}