{"record":{"id":"af0662b0251d25b8","repo":"pinpoint-apm/pinpoint","slug":"failed-to-handle-realtime-flux-item-sink-not-f","errorCode":null,"errorMessage":"Failed to handle realtime flux item: sink {} not found","messagePattern":"Failed to handle realtime flux item: 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":80,"sourceCode":"\n        logger.debug(\"Realtime flux item received: sinkId = {}\", sinkId);\n        if (!isHello) {\n            publisher.publish(response);\n        }\n    }\n\n    private ActiveThreadCountPublisher ensureSink(PCmdActiveThreadCountRes response) {\n        if (this.sinkId == -1 || publisher == null) {\n            return initSink(response);\n        }\n        return this.publisher;\n    }\n\n    private ActiveThreadCountPublisher initSink(PCmdActiveThreadCountRes response) {\n        this.sinkId = this.extractSinkId(response);\n        this.publisher = this.sinkRepository.get(sinkId);\n        if (this.publisher == null) {\n            logger.warn(\"Failed to handle realtime flux item: sink {} not found\", this.sinkId);\n            return null;\n        } else {\n            publisher.setStreamObserver(this.serverCallStreamObserver);\n        }\n        return publisher;\n    }\n\n    @Override\n    public void onError(Throwable t) {\n        final Status status = Status.fromThrowable(t);\n        if (Status.CANCELLED == status) {\n            logger.info(\"Stream cancelled: sinkId = {} {}\", sinkId, status);\n        } else {\n            logger.warn(\"Stream error: sinkId = {}, {}\", sinkId, status);\n        }\n\n        this.serverCallStreamObserver.onCompleted();\n","sourceCodeStart":62,"sourceCodeEnd":98,"githubUrl":"https://github.com/pinpoint-apm/pinpoint/blob/744c3d3075e595656abb1ae331ad2c0e4c9eb996/realtime/realtime-collector/src/main/java/com/navercorp/pinpoint/realtime/collector/receiver/grpc/ActiveThreadCountResponseStreamObserver.java#L62-L98","documentation":"Warn log in ActiveThreadCountResponseStreamObserver.initSink: the first realtime response arrived but its sinkId maps to no publisher in the sink repository, so the reactive stream has nowhere to publish — typically an expired/unknown sinkId or a first message that failed extraction.","triggerScenarios":"Thrown at realtime/realtime-collector/src/main/java/com/navercorp/pinpoint/realtime/collector/receiver/grpc/ActiveThreadCountResponseStreamObserver.java:80 when the library encounters an invalid state.","commonSituations":"See trigger scenarios.","solutions":["Verify the sinkId sent by the agent matches one issued by the collector","Check sink expiry/cleanup timing so streams outlive their sinks","Complete the gRPC stream with an error so the client can re-subscribe"],"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"}