private void updateListenersProgress(long highWatermark) {
        for (ListenerContext listenerContext : listenerContexts.values()) {
            listenerContext.nextExpectedOffset().ifPresent(nextExpectedOffset -> {
                // Send snapshot to the listener, if there is a snapshot for the partition,
                // and it is a new listener or
                // the listener is trying to read an offset for which there isn't a segment in the
                // log.
                if (nextExpectedOffset < highWatermark &&
                    (nextExpectedOffset == ListenerContext.STARTING_NEXT_OFFSET ||
                     nextExpectedOffset < log.startOffset()) &&
                    latestSnapshot().isPresent()
                ) {
                    listenerContext.fireHandleSnapshot(latestSnapshot().get());
                } else if (nextExpectedOffset == ListenerContext.STARTING_NEXT_OFFSET) {
                    // Reset the next offset to 0 since it is a new listener context and there are
                    // no checkpoint
                    logger.info(
                        "Setting the next offset of {} to {} since there are no snapshots",
                        listenerContext.listenerName(),
                        ListenerContext.SMALLEST_LOG_OFFSET
                    );
                    listenerContext.resetOffsetToSmallestLogOffset();
                } else if (nextExpectedOffset < log.startOffset()) {
                    throw new IllegalStateException(
                        String.format(
                            "Snapshot expected since next offset of %s is %d, log start offset " +
                            "is %d and high-watermark is %d",
                            listenerContext.listenerName(),
                            nextExpectedOffset,
                            log.startOffset(),
                            highWatermark
                        )
                    );
                }
            });

            // Re-read the expected offset in case the snapshot had to be reloaded
            listenerContext.nextExpectedOffset().ifPresent(nextExpectedOffset -> {
                if (nextExpectedOffset < highWatermark) {
                    LogFetchInfo readInfo = log.read(nextExpectedOffset, Isolation.COMMITTED);
                    listenerContext.fireHandleCommit(nextExpectedOffset, readInfo.records);
                }
            });
        }
    }
Create a new instance.

Note that if the node ID is empty, then the client will behave as a
non-participating observer.
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java