vvcephei commented on a change in pull request #8963:
URL: https://github.com/apache/kafka/pull/8963#discussion_r449041784
##########
File path:
streams/src/test/java/org/apache/kafka/streams/integration/EosBetaUpgradeIntegrationTest.java
##########
@@ -837,6 +866,11 @@ public void init(final ProcessorContext context) {
crash = errorInjectedClient2;
sharedCommit = commitCounterClient2;
}
+ punctuator = context.schedule(
+ Duration.ofSeconds(5),
Review comment:
Would it speed up the test to choose a smaller number here?
##########
File path:
streams/src/test/java/org/apache/kafka/streams/integration/EosBetaUpgradeIntegrationTest.java
##########
@@ -837,6 +866,11 @@ public void init(final ProcessorContext context) {
crash = errorInjectedClient2;
sharedCommit = commitCounterClient2;
}
+ punctuator = context.schedule(
Review comment:
There's going to be a separate punctuator per task, right? Does the test
account for this?
##########
File path:
streams/src/test/java/org/apache/kafka/streams/integration/EosBetaUpgradeIntegrationTest.java
##########
@@ -147,6 +152,25 @@
private final AtomicInteger commitCounterClient2 = new AtomicInteger(-1);
private final AtomicInteger commitRequested = new AtomicInteger(0);
+ private final AtomicBoolean requestCommit = new AtomicBoolean(false);
+ private static class CommitPunctuator implements Punctuator {
+ final ProcessorContext context;
+ final AtomicBoolean requestCommit;
+
+ public CommitPunctuator(final ProcessorContext context, final
AtomicBoolean requestCommit) {
+ this.context = context;
+ this.requestCommit = requestCommit;
+ }
+
+ @Override
+ public void punctuate(final long timestamp) {
+ if (requestCommit.get()) {
+ context.commit();
+ requestCommit.set(false);
+ }
Review comment:
There should never be multiple requests, right? If there were, a second
request might arrive between 168 and 169, violating the desired property. In
that case, we should grab a lock instead. As long as there's only one
requesting thread, and it always waits for the commit right after requesting,
then we should be good.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
