pinpoint-apm/pinpoint · error

Failed to handle realtime flux item: sink

Error message

Failed to handle realtime flux item: sink {} not found

What it means

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.

Solutions

1. Verify the sinkId sent by the agent matches one issued by the collector
2. Check sink expiry/cleanup timing so streams outlive their sinks
3. Complete the gRPC stream with an error so the client can re-subscribe

When it happens

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

Common situations: See trigger scenarios.

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

Appendix: source

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

        logger.debug("Realtime flux item received: sinkId = {}", sinkId);
        if (!isHello) {
            publisher.publish(response);
        }
    }

    private ActiveThreadCountPublisher ensureSink(PCmdActiveThreadCountRes response) {
        if (this.sinkId == -1 || publisher == null) {
            return initSink(response);
        }
        return this.publisher;
    }

    private ActiveThreadCountPublisher initSink(PCmdActiveThreadCountRes response) {
        this.sinkId = this.extractSinkId(response);
        this.publisher = this.sinkRepository.get(sinkId);
        if (this.publisher == null) {
            logger.warn("Failed to handle realtime flux item: sink {} not found", this.sinkId);
            return null;
        } else {
            publisher.setStreamObserver(this.serverCallStreamObserver);
        }
        return publisher;
    }

    @Override
    public void onError(Throwable t) {
        final Status status = Status.fromThrowable(t);
        if (Status.CANCELLED == status) {
            logger.info("Stream cancelled: sinkId = {} {}", sinkId, status);
        } else {
            logger.warn("Stream error: sinkId = {}, {}", sinkId, status);
        }

        this.serverCallStreamObserver.onCompleted();

View on GitHub (pinned to 744c3d3075)