private Map<TopicPartition, Long> checkpointableOffsets() {
        final Map<TopicPartition, Long> checkpointableOffsets = new HashMap<>(recordCollector.offsets());
        for (final Map.Entry<TopicPartition, Long> entry : consumedOffsets.entrySet()) {
            checkpointableOffsets.putIfAbsent(entry.getKey(), entry.getValue());
        }

        log.debug("Checkpointable offsets {}", checkpointableOffsets);

        return checkpointableOffsets;
    }
Return all the checkpointable offsets(written + consumed) to the state manager.
Currently only changelog topic offsets need to be checkpointed.
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java