KAFKA-19871: Add a TopologyTestDriverBuilder, deprecate old constructors

streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java

@@ -161,7 +161,9 @@
  * props.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
  * props.setProperty(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
  * Topology topology = ...
- * TopologyTestDriver driver = new TopologyTestDriver(topology, props);
+ * TopologyTestDriver driver = new TopologyTestDriverBuilder(topology)
+ *     .withConfig(props)
+ *     .build();
  * }</pre>
  *
  * <p> Note that the {@code TopologyTestDriver} processes input records synchronously.
@@ -265,45 +267,53 @@ public void onRestoreEnd(final TopicPartition topicPartition, final String store
     };
 
     /**
-     * Create a new test diver instance.
+     * Create a new test driver instance.
      * Default test properties are used to initialize the driver instance
      *
      * @param topology the topology to be tested
+     * @deprecated Since 4.4. Use {@link TopologyTestDriverBuilder} instead.
      */
+    @Deprecated(since = "4.4")
     public TopologyTestDriver(final Topology topology) {
         this(topology, new Properties());
     }
 
     /**
-     * Create a new test diver instance.
+     * Create a new test driver instance.
      * Initialized the internally mocked wall-clock time with {@link System#currentTimeMillis() current system time}.
      *
      * @param topology the topology to be tested
      * @param config   the configuration for the topology
+     * @deprecated Since 4.4. Use {@link TopologyTestDriverBuilder} instead.
      */
+    @Deprecated(since = "4.4")
     public TopologyTestDriver(final Topology topology,
                               final Properties config) {
         this(topology, config, null);
     }
 
     /**
-     * Create a new test diver instance.
+     * Create a new test driver instance.
      *
      * @param topology the topology to be tested
      * @param initialWallClockTimeMs the initial value of internally mocked wall-clock time
+     * @deprecated Since 4.4. Use {@link TopologyTestDriverBuilder} instead.
      */
+    @Deprecated(since = "4.4")
     public TopologyTestDriver(final Topology topology,
                               final Instant initialWallClockTimeMs) {
         this(topology, new Properties(), initialWallClockTimeMs);
     }
 
     /**
-     * Create a new test diver instance.
+     * Create a new test driver instance.
      *
      * @param topology               the topology to be tested
      * @param config                 the configuration for the topology
      * @param initialWallClockTime   the initial value of internally mocked wall-clock time
+     * @deprecated Since 4.4. Use {@link TopologyTestDriverBuilder} instead.
      */
+    @Deprecated(since = "4.4")
     public TopologyTestDriver(final Topology topology,
                               final Properties config,
                               final Instant initialWallClockTime) {
@@ -314,15 +324,16 @@ public TopologyTestDriver(final Topology topology,
     }
 
     /**
-     * Create a new test diver instance.
+     * Create a new test driver instance. Package-private core constructor shared by the (deprecated)
+     * public constructors and by {@link TopologyTestDriverBuilder}, which is the blessed entry point.
      *
      * @param builder builder for the topology to be tested
      * @param config the configuration for the topology
      * @param initialWallClockTimeMs the initial value of internally mocked wall-clock time
      */
-    private TopologyTestDriver(final InternalTopologyBuilder builder,
-                               final Properties config,
-                               final long initialWallClockTimeMs) {
+    TopologyTestDriver(final InternalTopologyBuilder builder,
+                       final Properties config,
+                       final long initialWallClockTimeMs) {
         final Properties configCopy = new Properties();
         configCopy.putAll(config);
         configCopy.putIfAbsent(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy-bootstrap-host:0");
@@ -450,7 +461,7 @@ private void setupGlobalTask(final Time mockWallClockTime,
 
             @SuppressWarnings("deprecation")
             final boolean globalEnabled = streamsConfig.getBoolean(StreamsConfig.PROCESSING_EXCEPTION_HANDLER_GLOBAL_ENABLED_CONFIG);
-            final ProcessingExceptionHandler processingExceptionHandler = 
+            final ProcessingExceptionHandler processingExceptionHandler =
                 globalEnabled ? streamsConfig.processingExceptionHandler() : null;
 
             globalStateTask = new GlobalStateUpdateTask(
@@ -837,7 +848,8 @@ <K, V> TestRecord<K, V> readRecord(final String topic,
         }
         final K key = keyDeserializer.deserialize(record.topic(), record.headers(), record.key());
         final V value = valueDeserializer.deserialize(record.topic(), record.headers(), record.value());
-        return new TestRecord<>(key, value, record.headers(), record.timestamp());
+        final int outputPartition = -1;
+        return new TestRecord<>(key, value, record.headers(), Instant.ofEpochMilli(record.timestamp()), outputPartition);
     }
 
     <K, V> void pipeRecord(final String topic,

streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriverBuilder.java

@@ -0,0 +1,88 @@
+/*
+ * 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.Instant;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+
+/**
+ * Fluent builder for a {@link TopologyTestDriver}.
+ *
+ * <p>This is the entry point for constructing a {@link TopologyTestDriver}.
+ * Configure the builder and call {@link #build()}.
+ * The {@link TopologyTestDriver} constructors remain functional but are deprecated in favor of
+ * this builder.</p>
+ *
+ * <pre>{@code
+ * TopologyTestDriver driver = new TopologyTestDriverBuilder(topology)
+ *     .withConfig(props)
+ *     .withInitialWallClockTime(Instant.ofEpochMilli(0))
+ *     .build();
+ * }</pre>
+ */
+public class TopologyTestDriverBuilder {
+
+    private final Topology topology;
+    private Properties config = new Properties();
+    private Optional<Instant> initialWallClockTime = Optional.empty();
+
+    /**
+     * Start building a driver for the given topology.
+     *
+     * @param topology the topology to be tested
+     */
+    public TopologyTestDriverBuilder(final Topology topology) {
+        this.topology = Objects.requireNonNull(topology, "topology cannot be null");
+    }
+
+    /**
+     * Set the configuration passed to the driver. Optional; defaults to empty {@link Properties}.
+     *
+     * @param config the configuration for the topology
+     * @return this builder
+     */
+    public TopologyTestDriverBuilder withConfig(final Properties config) {
+        this.config = Objects.requireNonNull(config, "config cannot be null");
+        return this;
+    }
+
+    /**
+     * Set the initial value of the driver's internally mocked wall-clock time. Optional; defaults to
+     * the current system time.
+     *
+     * @param initialWallClockTime the initial mocked wall-clock time
+     * @return this builder
+     */
+    public TopologyTestDriverBuilder withInitialWallClockTime(final Instant initialWallClockTime) {
+        this.initialWallClockTime = Optional.ofNullable(initialWallClockTime);
+        return this;
+    }
+
+    /**
+     * Build the driver: construct it and apply all declared topic partition counts.
+     *
+     * @return a ready-to-use {@link TopologyTestDriver}
+     */
+    public TopologyTestDriver build() {
+        return new TopologyTestDriver(
+            topology.internalTopologyBuilder,
+            config,
+            initialWallClockTime.map(Instant::toEpochMilli).orElseGet(System::currentTimeMillis));
+    }
+}