KAFKA-7277: Migrate Streams API to Duration instead of longMs times

streams/examples/src/main/java/org/apache/kafka/streams/examples/pageview/PageViewTypedDemo.java

@@ -19,6 +19,7 @@
 import com.fasterxml.jackson.annotation.JsonSubTypes;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import java.time.Duration;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.common.errors.SerializationException;
 import org.apache.kafka.common.serialization.Deserializer;
@@ -39,7 +40,6 @@
 import java.util.Map;
 import java.util.Properties;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
 
 /**
  * Demonstrates how to perform a join between a KStream and a KTable, i.e. an example of a stateful computation,
@@ -207,7 +207,7 @@ public static void main(final String[] args) {
             })
             .map((user, viewRegion) -> new KeyValue<>(viewRegion.region, viewRegion))
             .groupByKey(Serialized.with(Serdes.String(), new JSONSerde<>()))
-            .windowedBy(TimeWindows.of(TimeUnit.DAYS.toMillis(7)).advanceBy(TimeUnit.SECONDS.toMillis(1)))
+            .windowedBy(TimeWindows.of(Duration.ofDays(7)).advanceBy(Duration.ofSeconds(1)))
             .count()
             .toStream()
             .map((key, value) -> {

streams/examples/src/main/java/org/apache/kafka/streams/examples/pageview/PageViewUntypedDemo.java

@@ -19,6 +19,7 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.JsonNodeFactory;
 import com.fasterxml.jackson.databind.node.ObjectNode;
+import java.time.Duration;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.common.serialization.Deserializer;
 import org.apache.kafka.common.serialization.Serde;
@@ -87,7 +88,7 @@ public static void main(final String[] args) throws Exception {
             })
             .map((user, viewRegion) -> new KeyValue<>(viewRegion.get("region").textValue(), viewRegion))
             .groupByKey(Serialized.with(Serdes.String(), jsonSerde))
-            .windowedBy(TimeWindows.of(7 * 24 * 60 * 60 * 1000L).advanceBy(1000))
+            .windowedBy(TimeWindows.of(Duration.ofDays(7)).advanceBy(Duration.ofSeconds(1)))
             .count()
             .toStream()
             .map((key, value) -> {

streams/examples/src/main/java/org/apache/kafka/streams/examples/temperature/TemperatureDemo.java

@@ -16,6 +16,7 @@
  */
 package org.apache.kafka.streams.examples.temperature;
 
+import java.time.Duration;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.common.serialization.Serde;
 import org.apache.kafka.common.serialization.Serdes;
@@ -30,7 +31,6 @@
 
 import java.util.Properties;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
 
 /**
  * Demonstrates, using the high-level KStream DSL, how to implement an IoT demo application
@@ -88,7 +88,7 @@ public static void main(final String[] args) {
             // to group and reduce them, a key is needed ("temp" has been chosen)
             .selectKey((key, value) -> "temp")
             .groupByKey()
-            .windowedBy(TimeWindows.of(TimeUnit.SECONDS.toMillis(TEMPERATURE_WINDOW_SIZE)))
+            .windowedBy(TimeWindows.of(Duration.ofSeconds(TEMPERATURE_WINDOW_SIZE)))
             .reduce((value1, value2) -> {
                 if (Integer.parseInt(value1) > Integer.parseInt(value2))
                     return value1;

streams/examples/src/main/java/org/apache/kafka/streams/examples/wordcount/WordCountProcessorDemo.java

@@ -16,6 +16,7 @@
  */
 package org.apache.kafka.streams.examples.wordcount;
 
+import java.time.Duration;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.streams.KafkaStreams;
@@ -63,7 +64,7 @@ public Processor<String, String> get() {
                 @SuppressWarnings("unchecked")
                 public void init(final ProcessorContext context) {
                     this.context = context;
-                    this.context.schedule(1000, PunctuationType.STREAM_TIME, timestamp -> {
+                    this.context.schedule(Duration.ofMillis(1000), PunctuationType.STREAM_TIME, timestamp -> {
                         try (final KeyValueIterator<String, Integer> iter = kvStore.all()) {
                             System.out.println("----------- " + timestamp + " ----------- ");
 

streams/src/main/java/org/apache/kafka/streams/ApiUtils.java

@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Objects;
+
+/**
+ */
+public final class ApiUtils {
+    private ApiUtils() {
+    }
+
+    /**
+     * Validates that milliseconds from duration {@code d} can be retrieved.
+     * @param d Duration to check
+     * @param name Name of params for an error message.
+     */
+    public static void validateMillisecondDuration(final Duration d, final String name) {
+        try {
+            Objects.requireNonNull(d);
+
+            //noinspection ResultOfMethodCallIgnored
+            d.toMillis();
+        } catch (final NullPointerException e) {
+            throw new IllegalArgumentException(name + " shouldn't be null.", e);
+        } catch (final ArithmeticException e) {
+            throw new IllegalArgumentException(name + " can't be converted to milliseconds. " +  d +
+                " is negative or too big", e);
+        }
+    }
+
+    /**
+     * Validates that milliseconds from instant {@code i} can be retrieved.
+     * @param i Instant to check
+     * @param name Name of params for an error message.
+     */
+    public static void validateMillisecondInstant(final Instant i, final String name) {
+        try {
+            Objects.requireNonNull(i);
+
+            //noinspection ResultOfMethodCallIgnored
+            i.toEpochMilli();
+        } catch (final NullPointerException e) {
+            throw new IllegalArgumentException(name + " shouldn't be null.", e);
+        } catch (final ArithmeticException e) {
+            throw new IllegalArgumentException(name + " can't be converted to milliseconds. " +  i +
+                " is negative or too big", e);
+        }
+    }
+}

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

@@ -16,6 +16,7 @@
  */
 package org.apache.kafka.streams;
 
+import java.time.Duration;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.producer.KafkaProducer;
@@ -813,7 +814,7 @@ public void run() {
      * This will block until all threads have stopped.
      */
     public void close() {
-        close(DEFAULT_CLOSE_TIMEOUT, TimeUnit.SECONDS);
+        close(Duration.ofSeconds(DEFAULT_CLOSE_TIMEOUT));
     }
 
     /**
@@ -826,9 +827,26 @@ public void close() {
      * @return {@code true} if all threads were successfully stopped—{@code false} if the timeout was reached
      * before all threads stopped
      * Note that this method must not be called in the {@code onChange} callback of {@link StateListener}.
+     * @deprecated Use {@link #close(Duration)} instead
      */
+    @Deprecated
     public synchronized boolean close(final long timeout, final TimeUnit timeUnit) {
-        log.debug("Stopping Streams client with timeoutMillis = {} ms.", timeUnit.toMillis(timeout));
+        return close(Duration.ofMillis(timeUnit.toMillis(timeout)));
+    }
+
+    /**
+     * Shutdown this {@code KafkaStreams} by signaling all the threads to stop, and then wait up to the timeout for the
+     * threads to join.
+     * A {@code timeout} of 0 means to wait forever.
+     *
+     * @param timeout  how long to wait for the threads to shutdown
+     * @return {@code true} if all threads were successfully stopped—{@code false} if the timeout was reached
+     * before all threads stopped
+     * Note that this method must not be called in the {@code onChange} callback of {@link StateListener}.
+     * @throws IllegalArgumentException if {@param timeout} is negative or too big
+     */
+    public synchronized boolean close(final Duration timeout) throws IllegalArgumentException {
+        log.debug("Stopping Streams client with timeoutMillis = {} ms.", timeout.toMillis());
 
         if (!setState(State.PENDING_SHUTDOWN)) {
             // if transition failed, it means it was either in PENDING_SHUTDOWN
@@ -885,7 +903,7 @@ public void run() {
             shutdownThread.start();
         }
 
-        if (waitOnState(State.NOT_RUNNING, timeUnit.toMillis(timeout))) {
+        if (waitOnState(State.NOT_RUNNING, timeout.toMillis())) {
             log.info("Streams client stopped completely");
             return true;
         } else {

streams/src/main/java/org/apache/kafka/streams/kstream/JoinWindows.java

@@ -16,6 +16,7 @@
  */
 package org.apache.kafka.streams.kstream;
 
+import org.apache.kafka.streams.ApiUtils;
 import org.apache.kafka.streams.processor.TimestampExtractor;
 
 import java.time.Duration;
@@ -109,11 +110,27 @@ private JoinWindows(final long beforeMs,
      * the timestamp of the record from the primary stream.
      *
      * @param timeDifferenceMs join window interval in milliseconds
-     * @throws IllegalArgumentException if {@code timeDifferenceMs} is negative
+     * @throws IllegalArgumentException if {@code timeDifferenceMs} is negative or too big
+     * @deprecated Use {@link #of(Duration)} instead.
      */
+    @Deprecated
     public static JoinWindows of(final long timeDifferenceMs) throws IllegalArgumentException {
         // This is a static factory method, so we initialize grace and retention to the defaults.
-        return new JoinWindows(timeDifferenceMs, timeDifferenceMs, null, DEFAULT_RETENTION_MS);
+        return of(Duration.ofMillis(timeDifferenceMs));
+    }
+
+    /**
+     * Specifies that records of the same key are joinable if their timestamps are within {@code timeDifference},
+     * i.e., the timestamp of a record from the secondary stream is max {@code timeDifference} earlier or later than
+     * the timestamp of the record from the primary stream.
+     *
+     * @param timeDifference join window interval
+     * @throws IllegalArgumentException if {@code timeDifference} is negative or too big
+     */
+    public static JoinWindows of(final Duration timeDifference) throws IllegalArgumentException {
+        ApiUtils.validateMillisecondDuration(timeDifference, "timeDifference");
+
+        return new JoinWindows(timeDifference.toMillis(), timeDifference.toMillis(), null, DEFAULT_RETENTION_MS);
     }
 
     /**
@@ -124,11 +141,30 @@ public static JoinWindows of(final long timeDifferenceMs) throws IllegalArgument
      * value (which would result in a negative window size).
      *
      * @param timeDifferenceMs relative window start time in milliseconds
-     * @throws IllegalArgumentException if the resulting window size is negative
+     * @throws IllegalArgumentException if the resulting window size is negative or too big
+     * @deprecated Use {@link #before(Duration)} instead.
      */
     @SuppressWarnings({"deprecation"}) // removing segments from Windows will fix this
+    @Deprecated
     public JoinWindows before(final long timeDifferenceMs) throws IllegalArgumentException {
-        return new JoinWindows(timeDifferenceMs, afterMs, grace, maintainDurationMs, segments);
+        return before(Duration.ofMillis(timeDifferenceMs));
+    }
+
+    /**
+     * Changes the start window boundary to {@code timeDifference} but keep the end window boundary as is.
+     * Thus, records of the same key are joinable if the timestamp of a record from the secondary stream is at most
+     * {@code timeDifference} earlier than the timestamp of the record from the primary stream.
+     * {@code timeDifference} can be negative but it's absolute value must not be larger than current window "after"
+     * value (which would result in a negative window size).
+     *
+     * @param timeDifference relative window start time
+     * @throws IllegalArgumentException if the resulting window size is negative or too big
+     */
+    @SuppressWarnings({"deprecation"}) // removing segments from Windows will fix this
+    public JoinWindows before(final Duration timeDifference) throws IllegalArgumentException {
+        ApiUtils.validateMillisecondDuration(timeDifference, "timeDifference");
+
+        return new JoinWindows(timeDifference.toMillis(), afterMs, grace, maintainDurationMs, segments);
     }
 
     /**
@@ -139,11 +175,30 @@ public JoinWindows before(final long timeDifferenceMs) throws IllegalArgumentExc
      * value (which would result in a negative window size).
      *
      * @param timeDifferenceMs relative window end time in milliseconds
-     * @throws IllegalArgumentException if the resulting window size is negative
+     * @throws IllegalArgumentException if the resulting window size is negative or too big
+     * @deprecated Use {@link #after(Duration)} instead
      */
     @SuppressWarnings({"deprecation"}) // removing segments from Windows will fix this
+    @Deprecated
     public JoinWindows after(final long timeDifferenceMs) throws IllegalArgumentException {
-        return new JoinWindows(beforeMs, timeDifferenceMs, grace, maintainDurationMs, segments);
+        return after(Duration.ofMillis(timeDifferenceMs));
+    }
+
+    /**
+     * Changes the end window boundary to {@code timeDifference} but keep the start window boundary as is.
+     * Thus, records of the same key are joinable if the timestamp of a record from the secondary stream is at most
+     * {@code timeDifference} later than the timestamp of the record from the primary stream.
+     * {@code timeDifference} can be negative but it's absolute value must not be larger than current window "before"
+     * value (which would result in a negative window size).
+     *
+     * @param timeDifference relative window end time
+     * @throws IllegalArgumentException if the resulting window size is negative or too big
+     */
+    @SuppressWarnings({"deprecation"}) // removing segments from Windows will fix this
+    public JoinWindows after(final Duration timeDifference) throws IllegalArgumentException {
+        ApiUtils.validateMillisecondDuration(timeDifference, "timeDifference");
+
+        return new JoinWindows(beforeMs, timeDifference.toMillis(), grace, maintainDurationMs, segments);
     }
 
     /**
@@ -163,20 +218,19 @@ public long size() {
     }
 
     /**
-     * Reject late events that arrive more than {@code millisAfterWindowEnd}
+     * Reject late events that arrive more than {@code afterWindowEnd}
      * after the end of its window.
      *
      * Lateness is defined as (stream_time - record_timestamp).
      *
-     * @param millisAfterWindowEnd The grace period to admit late-arriving events to a window.
+     * @param afterWindowEnd The grace period to admit late-arriving events to a window.
      * @return this updated builder
      */
     @SuppressWarnings({"deprecation"}) // removing segments from Windows will fix this
-    public JoinWindows grace(final long millisAfterWindowEnd) {
-        if (millisAfterWindowEnd < 0) {
-            throw new IllegalArgumentException("Grace period must not be negative.");
-        }
-        return new JoinWindows(beforeMs, afterMs, Duration.ofMillis(millisAfterWindowEnd), maintainDurationMs, segments);
+    public JoinWindows grace(final Duration afterWindowEnd) throws IllegalArgumentException {
+        ApiUtils.validateMillisecondDuration(afterWindowEnd, "afterWindowEnd");
+
+        return new JoinWindows(beforeMs, afterMs, afterWindowEnd, maintainDurationMs, segments);
     }
 
     @SuppressWarnings("deprecation") // continuing to support Windows#maintainMs/segmentInterval in fallback mode
@@ -192,7 +246,7 @@ public long gracePeriodMs() {
      * @param durationMs the window retention time in milliseconds
      * @return itself
      * @throws IllegalArgumentException if {@code durationMs} is smaller than the window size
-     * @deprecated since 2.1. Use {@link JoinWindows#grace(long)} instead.
+     * @deprecated since 2.1. Use {@link JoinWindows#grace(Duration)} instead.
      */
     @SuppressWarnings("deprecation")
     @Override

streams/src/main/java/org/apache/kafka/streams/kstream/Materialized.java

@@ -19,6 +19,7 @@
 import org.apache.kafka.common.internals.Topic;
 import org.apache.kafka.common.serialization.Serde;
 import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.ApiUtils;
 import org.apache.kafka.streams.processor.StateStore;
 import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier;
 import org.apache.kafka.streams.state.KeyValueStore;
@@ -243,12 +244,12 @@ public Materialized<K, V, S> withCachingDisabled() {
      * from window-start through window-end, and for the entire grace period.
      *
      * @return itself
+     * @throws IllegalArgumentException if retention is negative or too big
      */
-    public Materialized<K, V, S> withRetention(final long retentionMs) {
-        if (retentionMs < 0) {
-            throw new IllegalArgumentException("Retention must not be negative.");
-        }
-        retention = Duration.ofMillis(retentionMs);
+    public Materialized<K, V, S> withRetention(final Duration retention) throws IllegalArgumentException {
+        ApiUtils.validateMillisecondDuration(retention, "retention");
+
+        this.retention = retention;
         return this;
     }
 }