pinpoint-apm/pinpoint · error

Failed to emit complete: sink not found

Error message

Failed to emit complete: sink not found

What it means

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.

Solutions

1. Ensure the first response carries a valid sinkId so initSink binds a publisher
2. Guard subscribers with a timeout so they fail rather than hang
3. Log the sinkId (if any) to correlate with the issuing request

When it happens

Trigger: 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.

Common situations: See trigger scenarios.

AI-assisted analysis of pinpoint-apm/pinpoint@744c3d3075 (2026-09-07). Data as JSON: /api/errors/b25dd8b2f7ea7ee3. Report an issue: GitHub.

Appendix: source

Thrown at realtime/realtime-collector/src/main/java/com/navercorp/pinpoint/realtime/collector/receiver/grpc/ActiveThreadCountResponseStreamObserver.java:115

        this.serverCallStreamObserver.onCompleted();

        if (this.publisher != null) {
            this.publisher.error(t);
        } else {
            logger.warn("Failed to emit error: sink not found. the error may have occurred before the first message");
        }
    }

    @Override
    public void onCompleted() {
        logger.info("Completed stream: sinkId = {}", this.sinkId);

        this.serverCallStreamObserver.onCompleted();

        if (this.publisher != null) {
            this.publisher.complete();
        } else {
            logger.warn("Failed to emit complete: sink not found");
        }
    }

    private long extractSinkId(PCmdActiveThreadCountRes response) {
        return response.getCommonStreamResponse().getResponseId();
    }

    private int extractSequence(PCmdActiveThreadCountRes response) {
        return response.getCommonStreamResponse().getSequenceId();
    }

}

View on GitHub (pinned to 744c3d3075)