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

Code before:
        }
    }

    @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 recordContext = new ProcessorRecordContext(
                    time.milliseconds(),
                    -1L,
                    -1,
                    null,
                    new RecordHeaders()
            );
            updateProcessorContext(node, time.milliseconds(), recordContext);
            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
    public void addRecords(final TopicPartition partition, final Iterable<ConsumerRecord<byte[], byte[]>> records) {
        final int newQueueSize = partitionGroup.addRawRecords(partition, records);


Review comment (suggestion):
Not sure if this would be what we want? We would set a single TS, and if `init()` runs for a longer period of time, all records would get the same TS?

We might need to change `ProcessorRecordContext` to have a "time function" instead of a hard-coded value? For a record at hand, the time function would be `() -> record.timestamp()` and return the same value each time, and for this dummy context, we would use `() -> time.milliseconds()`.

But need to think about this more, and figure out what design we really want to get to, and what the splash radius of changing `ProcessorRecordContext` would be :) 

Code after:
                // 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
        if (newQueueSize > maxBufferedSize) {
            mainConsumer.pause(singleton(partition));
        }
    }

    public void updateNextOffsets(final TopicPartition partition, final OffsetAndMetadata offsetAndMetadata) {
        nextOffsetsAndMetadataToBeConsumed.put(partition, offsetAndMetadata);
    }