Set time for embedded cluster to current time and set low timestamps …

…to current time + timestamp
apache · Sep 19, 2024 · 74dc218 · 74dc218
1 parent f3aa2fd
commit 74dc218
Show file tree

Hide file tree

Showing 5 changed files with 31 additions and 25 deletions.
diff --git a/...t/java/org/apache/kafka/streams/integration/JoinGracePeriodDurabilityIntegrationTest.java b/...t/java/org/apache/kafka/streams/integration/JoinGracePeriodDurabilityIntegrationTest.java
@@ -38,7 +38,6 @@
 import org.apache.kafka.streams.state.Stores;
 import org.apache.kafka.test.MockValueJoiner;
 import org.apache.kafka.test.TestUtils;
-
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Tag;
@@ -49,6 +48,7 @@
 
 import java.io.IOException;
 import java.time.Duration;
+import java.time.Instant;
 import java.util.List;
 import java.util.Optional;
 import java.util.Properties;
@@ -70,10 +70,11 @@
 @Timeout(600)
 public class JoinGracePeriodDurabilityIntegrationTest {
 
+    private static final long NOW = Instant.now().toEpochMilli();
+
     public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(
         3,
-        new Properties(),
-        0L
+        new Properties()
     );
 
     @BeforeAll
@@ -219,7 +220,7 @@ private void verifyOutput(final String topic, final List<KeyValueTimestamp<Strin
      * just to exercise that everything works properly in the presence of commits.
      */
     private long scaledTime(final long unscaledTime) {
-        return COMMIT_INTERVAL * 2 * unscaledTime;
+        return NOW + COMMIT_INTERVAL * 2 * unscaledTime;
     }
 
     private static void produceSynchronouslyToPartitionZero(final String topic, final List<KeyValueTimestamp<String, String>> toProduce) {

diff --git a/...src/test/java/org/apache/kafka/streams/integration/ResetPartitionTimeIntegrationTest.java b/...src/test/java/org/apache/kafka/streams/integration/ResetPartitionTimeIntegrationTest.java
@@ -44,6 +44,7 @@
 import org.junit.jupiter.params.provider.ValueSource;
 
 import java.io.IOException;
+import java.time.Instant;
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
@@ -63,14 +64,15 @@
 @Timeout(600)
 public class ResetPartitionTimeIntegrationTest {
     private static final int NUM_BROKERS = 1;
+    private static final long NOW = Instant.now().toEpochMilli();
     private static final Properties BROKER_CONFIG;
     static {
         BROKER_CONFIG = new Properties();
         BROKER_CONFIG.put("transaction.state.log.replication.factor", (short) 1);
         BROKER_CONFIG.put("transaction.state.log.min.isr", 1);
     }
     public static final EmbeddedKafkaCluster CLUSTER =
-        new EmbeddedKafkaCluster(NUM_BROKERS, BROKER_CONFIG, 0L);
+        new EmbeddedKafkaCluster(NUM_BROKERS, BROKER_CONFIG);
 
     @BeforeAll
     public static void startCluster() throws IOException {
@@ -118,13 +120,13 @@ public void shouldPreservePartitionTimeOnKafkaStreamRestart(final String process
             produceSynchronouslyToPartitionZero(
                 input,
                 Collections.singletonList(
-                    new KeyValueTimestamp<>("k3", "v3", 5000)
+                    new KeyValueTimestamp<>("k3", "v3", NOW + 5000)
                 )
             );
             verifyOutput(
                 outputRaw,
                 Collections.singletonList(
-                    new KeyValueTimestamp<>("k3", "v3", 5000)
+                    new KeyValueTimestamp<>("k3", "v3", NOW + 5000)
                 )
             );
             assertThat(lastRecordedTimestamp, is(-1L));
@@ -139,16 +141,16 @@ public void shouldPreservePartitionTimeOnKafkaStreamRestart(final String process
             produceSynchronouslyToPartitionZero(
                 input,
                 Collections.singletonList(
-                    new KeyValueTimestamp<>("k5", "v5", 4999)
+                    new KeyValueTimestamp<>("k5", "v5", NOW + 4999)
                 )
             );
             verifyOutput(
                 outputRaw,
                 Collections.singletonList(
-                    new KeyValueTimestamp<>("k5", "v5", 4999)
+                    new KeyValueTimestamp<>("k5", "v5", NOW + 4999)
                 )
             );
-            assertThat(lastRecordedTimestamp, is(5000L));
+            assertThat(lastRecordedTimestamp, is(NOW + 5000L));
         } finally {
             kafkaStreams.close();
             quietlyCleanStateAfterTest(CLUSTER, kafkaStreams);

diff --git a/.../org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java b/.../org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java
@@ -54,6 +54,7 @@
 
 import java.io.IOException;
 import java.time.Duration;
+import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -81,7 +82,8 @@
 @Timeout(600)
 public class StreamsUncaughtExceptionHandlerIntegrationTest {
 
-    public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(1, new Properties(), Collections.emptyMap(), 0L, 0L);
+    private static final long NOW = Instant.now().toEpochMilli();
+    public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(1, new Properties(), Collections.emptyMap());
 
     @BeforeAll
     public static void startCluster() throws IOException {
@@ -146,7 +148,7 @@ public void shouldShutdownThreadUsingOldHandler() throws Exception {
             kafkaStreams.setUncaughtExceptionHandler((t, e) -> counter.incrementAndGet());
 
             startApplicationAndWaitUntilRunning(kafkaStreams);
-            produceMessages(0L, inputTopic, "A");
+            produceMessages(NOW, inputTopic, "A");
 
             // should call the UncaughtExceptionHandler in current thread
             TestUtils.waitForCondition(() -> counter.get() == 1, "Handler was called 1st time");
@@ -168,7 +170,7 @@ public void shouldShutdownClient() throws Exception {
 
             startApplicationAndWaitUntilRunning(kafkaStreams);
 
-            produceMessages(0L, inputTopic, "A");
+            produceMessages(NOW, inputTopic, "A");
             waitForApplicationState(Collections.singletonList(kafkaStreams), KafkaStreams.State.ERROR, DEFAULT_DURATION);
 
             assertThat(processorValueCollector.size(), equalTo(1));
@@ -252,7 +254,7 @@ public void shouldShutDownClientIfGlobalStreamThreadWantsToReplaceThread() throw
 
             startApplicationAndWaitUntilRunning(kafkaStreams);
 
-            produceMessages(0L, inputTopic2, "A");
+            produceMessages(NOW, inputTopic2, "A");
             waitForApplicationState(Collections.singletonList(kafkaStreams), KafkaStreams.State.ERROR, DEFAULT_DURATION);
 
             assertThat(processorValueCollector.size(), equalTo(1));
@@ -297,7 +299,7 @@ public void shouldEmitSameRecordAfterFailover() throws Exception {
                     IntegerSerializer.class,
                     StringSerializer.class,
                     new Properties()),
-                0L);
+                NOW);
 
             IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
                 inputTopic2,
@@ -310,7 +312,7 @@ public void shouldEmitSameRecordAfterFailover() throws Exception {
                     IntegerSerializer.class,
                     StringSerializer.class,
                     new Properties()),
-                0L);
+                NOW);
 
             IntegrationTestUtils.waitUntilFinalKeyValueRecordsReceived(
                 TestUtils.consumerConfig(
@@ -365,7 +367,7 @@ private void testShutdownApplication(final int numThreads) throws Exception {
 
             startApplicationAndWaitUntilRunning(asList(kafkaStreams1, kafkaStreams2));
 
-            produceMessages(0L, inputTopic, "A");
+            produceMessages(NOW, inputTopic, "A");
             waitForApplicationState(asList(kafkaStreams1, kafkaStreams2), KafkaStreams.State.ERROR, DEFAULT_DURATION);
 
             assertThat(processorValueCollector.size(), equalTo(1));
@@ -386,7 +388,7 @@ private void testReplaceThreads(final int numThreads) throws Exception {
             });
             startApplicationAndWaitUntilRunning(kafkaStreams);
 
-            produceMessages(0L, inputTopic, "A");
+            produceMessages(NOW, inputTopic, "A");
             TestUtils.waitForCondition(() -> count.get() == numThreads, "finished replacing threads");
             TestUtils.waitForCondition(() -> throwError.get(), "finished replacing threads");
             kafkaStreams.close();

diff --git a/.../test/java/org/apache/kafka/streams/integration/SuppressionDurabilityIntegrationTest.java b/.../test/java/org/apache/kafka/streams/integration/SuppressionDurabilityIntegrationTest.java
@@ -55,6 +55,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.time.Instant;
 import java.util.List;
 import java.util.Optional;
 import java.util.Properties;
@@ -80,11 +81,10 @@
 @Tag("integration")
 @Timeout(600)
 public class SuppressionDurabilityIntegrationTest {
-
+    private static final long NOW = Instant.now().toEpochMilli();
     public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(
         3,
-        mkProperties(mkMap()),
-        0L
+        mkProperties(mkMap())
     );
 
     @BeforeAll
@@ -307,7 +307,7 @@ private void verifyOutput(final String topic, final List<KeyValueTimestamp<Strin
      * just to exercise that everything works properly in the presence of commits.
      */
     private long scaledTime(final long unscaledTime) {
-        return COMMIT_INTERVAL * 2 * unscaledTime;
+        return NOW + COMMIT_INTERVAL * 2 * unscaledTime;
     }
 
     private static void produceSynchronouslyToPartitionZero(final String topic, final List<KeyValueTimestamp<String, String>> toProduce) {

diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/SuppressionIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/SuppressionIntegrationTest.java
@@ -52,6 +52,7 @@
 import org.junit.jupiter.api.Timeout;
 
 import java.io.IOException;
+import java.time.Instant;
 import java.util.Collections;
 import java.util.List;
 import java.util.Locale;
@@ -83,11 +84,11 @@
 @Tag("integration")
 @Timeout(600)
 public class SuppressionIntegrationTest {
+    private static final long NOW = Instant.now().toEpochMilli();
 
     public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(
         1,
-        mkProperties(mkMap()),
-        0L
+        mkProperties(mkMap())
     );
 
     @BeforeAll
@@ -525,7 +526,7 @@ private static Properties getStreamsConfig(final String appId) {
      * just to exercise that everything works properly in the presence of commits.
      */
     private static long scaledTime(final long unscaledTime) {
-        return COMMIT_INTERVAL * 2 * unscaledTime;
+        return NOW + COMMIT_INTERVAL * 2 * unscaledTime;
     }
 
     private static void produceSynchronously(final String topic, final List<KeyValueTimestamp<String, String>> toProduce) {