Skip to content

KAFKA-6748: double check before scheduling a new task after the punctuate call #4827

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
guozhangwang merged 3 commits into from
Apr 5, 2018
Merged

KAFKA-6748: double check before scheduling a new task after the punctuate call #4827

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
1938db4
Test case to illustrate KAFKA-6748
Apr 5, 2018
d3cd12d
fix for KAFKA-6748
Apr 5, 2018
4f35351
Cleanup test
Apr 5, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion streams/src/main/java/org/apache/kafka/streams/processor/internals/PunctuationQueue.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,10 @@ boolean mayPunctuate(final long timestamp, final PunctuationType type, final Pro

if (!sched.isCancelled()) {
processorNodePunctuator.punctuate(sched.node(), timestamp, type, sched.punctuator());
pq.add(sched.next(timestamp));
// sched can be cancelled from within the punctuator
if (!sched.isCancelled()) {
pq.add(sched.next(timestamp));
}
punctuated = true;
}

Expand Down
37 changes: 37 additions & 0 deletions streams/src/test/java/org/apache/kafka/streams/processor/internals/PunctuationQueueTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
package org.apache.kafka.streams.processor.internals;

import org.apache.kafka.streams.processor.AbstractProcessor;
import org.apache.kafka.streams.processor.Cancellable;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.PunctuationType;
import org.apache.kafka.streams.processor.Punctuator;
Expand Down Expand Up @@ -126,6 +127,42 @@ public void punctuate(ProcessorNode node, long time, PunctuationType type, Punct
assertEquals(4, processor.punctuatedAt.size());
}

@Test
public void testPunctuationIntervalCancelFromPunctuator() {
final TestProcessor processor = new TestProcessor();
final ProcessorNode<String, String> node = new ProcessorNode<>("test", processor, null);
final PunctuationQueue queue = new PunctuationQueue();
final Punctuator punctuator = new Punctuator() {
@Override
public void punctuate(long timestamp) {
node.processor().punctuate(timestamp);
}
};

final PunctuationSchedule sched = new PunctuationSchedule(node, 0L, 100L, punctuator);
final long now = sched.timestamp - 100L;

final Cancellable cancellable = queue.schedule(sched);

ProcessorNodePunctuator processorNodePunctuator = new ProcessorNodePunctuator() {
@Override
public void punctuate(ProcessorNode node, long time, PunctuationType type, Punctuator punctuator) {
punctuator.punctuate(time);
// simulate scheduler cancelled from within punctuator
cancellable.cancel();
}
};

queue.mayPunctuate(now, PunctuationType.STREAM_TIME, processorNodePunctuator);
assertEquals(0, processor.punctuatedAt.size());

queue.mayPunctuate(now + 100L, PunctuationType.STREAM_TIME, processorNodePunctuator);
assertEquals(1, processor.punctuatedAt.size());

queue.mayPunctuate(now + 200L, PunctuationType.STREAM_TIME, processorNodePunctuator);
assertEquals(1, processor.punctuatedAt.size());
}

private static class TestProcessor extends AbstractProcessor<String, String> {

public final ArrayList<Long> punctuatedAt = new ArrayList<>();
Expand Down