vvcephei commented on a change in pull request #11213:
URL: https://github.com/apache/kafka/pull/11213#discussion_r703626660
##########
File path:
streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessorSupplier.java
##########
@@ -150,16 +153,16 @@ public void init(final
org.apache.kafka.streams.processor.ProcessorContext conte
}
@Override
- public void process(final K key, final Change<V> value) {
+ public void process(final Record<K, Change<V>> record) {
observedStreamTime = Math.max(observedStreamTime,
internalProcessorContext.timestamp());
- buffer(key, value);
+ buffer(record);
enforceConstraints();
}
- private void buffer(final K key, final Change<V> value) {
- final long bufferTime =
bufferTimeDefinition.time(internalProcessorContext, key);
+ private void buffer(final Record<K, Change<V>> record) {
+ final long bufferTime =
bufferTimeDefinition.time(internalProcessorContext, record.key());
- buffer.put(bufferTime, key, value,
internalProcessorContext.recordContext());
+ buffer.put(bufferTime, record.key(), record.value(),
internalProcessorContext.recordContext());
Review comment:
Thanks, @jeqo , I can see that my tone in that comment didn't encompass
the scope of the change I was proposing.
I'm fine if we want to delay that refactor for later work and focus on the
Processor API itself in this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
