Repository: apache/kafka
File: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java (Java)

Code before:
                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);
        }
    }

    @Override
    public Map<TopicPartition, Long> purgeableOffsets() {
        final Map<TopicPartition, Long> purgeableConsumedOffsets = new HashMap<>();
        for (final Map.Entry<TopicPartition, Long> entry : committedOffsets.entrySet()) {
            final TopicPartition tp = entry.getKey();
            if (topology.isRepartitionTopic(tp.topic())) {
                // committedOffsets map is initialized at -1 so no purging until there's a committed offset
                if (entry.getValue() > -1) {
                    purgeableConsumedOffsets.put(tp, entry.getValue());
                }
            }
        }

        return purgeableConsumedOffsets;
    }

    @SuppressWarnings("unchecked")
    private void initializeTopology() {
        // initialize the task by initializing all its processor nodes in the topology
        log.trace("Initializing processor nodes of the topology");
        for (final ProcessorNode<?, ?, ?, ?> node : topology.processors()) {
            final InitProcessorRecordContext initContext = new InitProcessorRecordContext(time);
            updateProcessorContext(node, time.milliseconds(), initContext);
            try {
                node.init(processorContext, processingExceptionHandler);
            } finally {
                updateProcessorContext(null, RecordQueue.UNKNOWN, null);
            }
        }
    }

    /**
     * Adds records to queues. If a record has an invalid (i.e., negative) timestamp, the record is skipped
     * and not added to the queue for processing
     *
     * @param partition the partition
     * @param records   the records
     */
    @Override

Review comment (performance):
Looking into `updateProcessorContext()` it does set "system time" to a fixed value. We did discuss this originally, if it make sense to have a static ts, or a moving one, and though we want a moving one.

But if system time is not moving either, I am wondering if we really need it, or should fall back to just getting system time once, and set on the context. 🤔 

Or would we also need to update `ProcessorRecordContext` to return a "dynamic" system time? We did opt to return a static one, as it is not for free to call `System.currentTimeMillis()` in general, so we want to cache the result and reuse to avoid unnecessary overhead. If this was "good enough" so far, it might also be "good enough" during initialization, too?

Thoughts? (Sorry for the back and forth...)

Code after:
        }
    }

    @Override
    public Map<TopicPartition, Long> purgeableOffsets() {
        final Map<TopicPartition, Long> purgeableConsumedOffsets = new HashMap<>();
        for (final Map.Entry<TopicPartition, Long> entry : committedOffsets.entrySet()) {
            final TopicPartition tp = entry.getKey();
            if (topology.isRepartitionTopic(tp.topic())) {
                // committedOffsets map is initialized at -1 so no purging until there's a committed offset
                if (entry.getValue() > -1) {
                    purgeableConsumedOffsets.put(tp, entry.getValue());
                }
            }
        }

        return purgeableConsumedOffsets;
    }

    @SuppressWarnings("unchecked")
    private void initializeTopology() {
        // initialize the task by initializing all its processor nodes in the topology
        log.trace("Initializing processor nodes of the topology");
        for (final ProcessorNode<?, ?, ?, ?> node : topology.processors()) {
            final ProcessorRecordContext initContext = new ProcessorRecordContext(time.milliseconds(), NO_OFFSET, NO_PARTITION, null, new RecordHeaders());
            updateProcessorContext(node, time.milliseconds(), initContext);
            try {
                node.init(processorContext, processingExceptionHandler);
            } finally {
                updateProcessorContext(null, ConsumerRecord.NO_TIMESTAMP, null);
            }
        }
    }

    /**
     * Adds records to queues. If a record has an invalid (i.e., negative) timestamp, the record is skipped
     * and not added to the queue for processing
     *
     * @param partition the partition
     * @param records   the records
     */
    @Override
    public void addRecords(final TopicPartition partition, final Iterable<ConsumerRecord<byte[], byte[]>> records) {
        final int newQueueSize = partitionGroup.addRawRecords(partition, records);

        if (log.isTraceEnabled()) {
            log.trace("Added records into the buffered queue of partition {}, new queue size is {}", partition, newQueueSize);
        }

        // if after adding these records, its partition queue's buffered size has been
        // increased beyond the threshold, we can then pause the consumption for this partition