private void initializeTaskTimeAndProcessorMetadata(final Map<TopicPartition, OffsetAndMetadata> offsetsAndMetadata) {
        final ProcessorMetadata finalProcessMetadata = new ProcessorMetadata();
        for (final Map.Entry<TopicPartition, OffsetAndMetadata> entry : offsetsAndMetadata.entrySet()) {
            final TopicPartition partition = entry.getKey();
            final OffsetAndMetadata metadata = entry.getValue();

            if (metadata != null) {
                final TopicPartitionMetadata committedTimestampAndMeta = TopicPartitionMetadata.decode(metadata.metadata());
                final long committedTimestamp = committedTimestampAndMeta.partitionTime();
                partitionGroup.setPartitionTime(partition, committedTimestamp);
                log.debug("A committed timestamp was detected: setting the partition time of partition {}"
                    + " to {} in stream task {}", partition, committedTimestamp, id);

                final ProcessorMetadata processorMetadata = committedTimestampAndMeta.processorMetadata();
                finalProcessMetadata.update(processorMetadata);
            } else {
                log.debug("No committed timestamp was found in metadata for partition {}", partition);
            }
        }
        processorContext.setProcessorMetadata(finalProcessMetadata);

        final Set<TopicPartition> nonCommitted = new HashSet<>(inputPartitions());
        nonCommitted.removeAll(offsetsAndMetadata.keySet());
        for (final TopicPartition partition : nonCommitted) {
            log.debug("No committed offset for partition {}, therefore no timestamp can be found for this partition", partition);
        }
    }
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