Skip to content

KAFKA-20903: Revert public API changes of incomplete KIP-1238 (Multi-Partition TDD) #23130

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

Open
mjsax wants to merge 3 commits into
base: 4.4
Choose a base branch
from
Open

KAFKA-20903: Revert public API changes of incomplete KIP-1238 (Multi-Partition TDD) #23130

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
4c1b3c0
Revert "KAFKA-19871: Update test with new TopologyTestDriverBuilder (…
Aug 11, 2026
ff47790
Revert "KAFKA-19871: Add a TopologyTestDriverBuilder, deprecate old c…
Aug 11, 2026
d0e845a
Revert "KAFKA-19871 - Add partition support to TestRecord (#22612)"
Aug 11, 2026
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
4 changes: 2 additions & 2 deletions docs/streams/developer-guide/testing.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,7 +60,7 @@ The test-utils package provides a `TopologyTestDriver` that can be used pipe dat
Topology topology = builder.build();

// create test driver
TopologyTestDriver testDriver = new TopologyTestDriverBuilder(topology).build();
TopologyTestDriver testDriver = new TopologyTestDriver(topology);

With the test driver you can create `TestInputTopic` giving topic name and the corresponding serializers. `TestInputTopic` provides various methods to pipe new message values, keys and values, or list of KeyValue objects.

Expand Down Expand Up @@ -119,7 +119,7 @@ The following example demonstrates how to use the test driver and helper classes
Properties props = new Properties();
props.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
props.setProperty(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.Long().getClass().getName());
testDriver = new TopologyTestDriverBuilder(topology).withConfig(props).build();
testDriver = new TopologyTestDriver(topology, props);

// setup test topics
inputTopic = testDriver.createInputTopic("input-topic", stringSerde.serializer(), longSerde.serializer());
Expand Down
3 changes: 1 addition & 2 deletions .../examples/src/test/java/org/apache/kafka/streams/examples/docs/DeveloperGuideTesting.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,6 @@
import org.apache.kafka.streams.TestOutputTopic;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.TopologyTestDriver;
import org.apache.kafka.streams.TopologyTestDriverBuilder;
import org.apache.kafka.streams.processor.PunctuationType;
import org.apache.kafka.streams.processor.api.Processor;
import org.apache.kafka.streams.processor.api.ProcessorContext;
Expand Down Expand Up @@ -76,7 +75,7 @@ public void setup() {
final Properties props = new Properties();
props.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class.getName());
props.setProperty(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.LongSerde.class.getName());
testDriver = new TopologyTestDriverBuilder(topology).withConfig(props).build();
testDriver = new TopologyTestDriver(topology, props);

// setup test topics
inputTopic = testDriver.createInputTopic("input-topic", stringSerde.serializer(), longSerde.serializer());
Expand Down
3 changes: 1 addition & 2 deletions ...examples/src/test/java/org/apache/kafka/streams/examples/wordcount/WordCountDemoTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,6 @@
import org.apache.kafka.streams.TestInputTopic;
import org.apache.kafka.streams.TestOutputTopic;
import org.apache.kafka.streams.TopologyTestDriver;
import org.apache.kafka.streams.TopologyTestDriverBuilder;
import org.apache.kafka.test.TestUtils;

import org.junit.jupiter.api.AfterEach;
Expand Down Expand Up @@ -59,7 +58,7 @@ public void setup() throws IOException {
final StreamsBuilder builder = new StreamsBuilder();
//Create Actual Stream Processing pipeline
WordCountDemo.createWordCountStream(builder);
testDriver = new TopologyTestDriverBuilder(builder.build()).withConfig(WordCountDemo.streamsConfig(null)).build();
testDriver = new TopologyTestDriver(builder.build(), WordCountDemo.streamsConfig(null));
inputTopic = testDriver.createInputTopic(WordCountDemo.INPUT_TOPIC, new StringSerializer(), new StringSerializer());
outputTopic = testDriver.createOutputTopic(WordCountDemo.OUTPUT_TOPIC, new StringDeserializer(), new LongDeserializer());
}
Expand Down
7 changes: 3 additions & 4 deletions ...tests/src/test/java/org/apache/kafka/streams/integration/AbstractJoinIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,6 @@
import org.apache.kafka.streams.TestOutputTopic;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.TopologyTestDriver;
import org.apache.kafka.streams.TopologyTestDriverBuilder;
import org.apache.kafka.streams.kstream.ValueJoiner;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
Expand Down Expand Up @@ -160,7 +159,7 @@ void runTestWithDriver(
final String storeName,
final Properties properties,
final Topology topology) {
try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(topology).withConfig(properties).build()) {
try (final TopologyTestDriver driver = new TopologyTestDriver(topology, properties)) {
final TestInputTopic<Long, String> right = driver.createInputTopic(INPUT_TOPIC_RIGHT, new LongSerializer(), new StringSerializer());
final TestInputTopic<Long, String> left = driver.createInputTopic(INPUT_TOPIC_LEFT, new LongSerializer(), new StringSerializer());
final TestOutputTopic<Long, String> outputTopic = driver.createOutputTopic(OUTPUT_TOPIC, new LongDeserializer(), new StringDeserializer());
Expand Down Expand Up @@ -204,7 +203,7 @@ void runTestWithDriver(
final String storeName,
final Properties streamsConfig,
final Topology topology) {
try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(topology).withConfig(streamsConfig).build()) {
try (final TopologyTestDriver driver = new TopologyTestDriver(topology, streamsConfig)) {
final TestInputTopic<Long, String> right = driver.createInputTopic(INPUT_TOPIC_RIGHT, new LongSerializer(), new StringSerializer());
final TestInputTopic<Long, String> left = driver.createInputTopic(INPUT_TOPIC_LEFT, new LongSerializer(), new StringSerializer());
final TestOutputTopic<Long, String> outputTopic = driver.createOutputTopic(OUTPUT_TOPIC, new LongDeserializer(), new StringDeserializer());
Expand Down Expand Up @@ -240,7 +239,7 @@ void runSelfJoinTestWithDriver(
final List<List<TestRecord<Long, String>>> expectedResult,
final Properties streamsConfig,
final Topology topology) {
try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(topology).withConfig(streamsConfig).build()) {
try (final TopologyTestDriver driver = new TopologyTestDriver(topology, streamsConfig)) {
final TestInputTopic<Long, String> left = driver.createInputTopic(INPUT_TOPIC_LEFT, new LongSerializer(), new StringSerializer());
final TestOutputTopic<Long, String> outputTopic = driver.createOutputTopic(OUTPUT_TOPIC, new LongDeserializer(), new StringDeserializer());

Expand Down
3 changes: 1 addition & 2 deletions ...sts/src/test/java/org/apache/kafka/streams/integration/KTableEfficientRangeQueryTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,6 @@
import org.apache.kafka.streams.TestInputTopic;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.TopologyTestDriver;
import org.apache.kafka.streams.TopologyTestDriverBuilder;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier;
import org.apache.kafka.streams.state.KeyValueIterator;
Expand Down Expand Up @@ -120,7 +119,7 @@ public void testStoreConfig(final StoreType storeType, final boolean enableLoggi
builder.table("input", stateStoreConfig);
final Topology topology = builder.build();

try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(topology).build()) {
try (final TopologyTestDriver driver = new TopologyTestDriver(topology)) {
//get input topic and stateStore
final TestInputTopic<String, String> input = driver
.createInputTopic("input", new StringSerializer(), new StringSerializer());
Expand Down
21 changes: 10 additions & 11 deletions .../java/org/apache/kafka/streams/integration/KTableKTableForeignKeyJoinIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,6 @@
import org.apache.kafka.streams.TestOutputTopic;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.TopologyTestDriver;
import org.apache.kafka.streams.TopologyTestDriverBuilder;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
Expand Down Expand Up @@ -237,7 +236,7 @@ public void doJoinFromLeftThenDeleteLeftEntity(final boolean leftJoin,
final boolean transactional) {
final Properties streamsConfig = getStreamsProperties(optimization, withHeaders, transactional);
final Topology topology = getTopology(streamsConfig, materialized ? "store" : null, leftJoin, rejoin, leftVersioned, rightVersioned);
try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(topology).withConfig(streamsConfig).build()) {
try (final TopologyTestDriver driver = new TopologyTestDriver(topology, streamsConfig)) {
final TestInputTopic<String, String> right = driver.createInputTopic(RIGHT_TABLE, new StringSerializer(), new StringSerializer());
final TestInputTopic<String, String> left = driver.createInputTopic(LEFT_TABLE, new StringSerializer(), new StringSerializer());
final TestOutputTopic<String, String> outputTopic = driver.createOutputTopic(OUTPUT, new StringDeserializer(), new StringDeserializer());
Expand Down Expand Up @@ -363,7 +362,7 @@ public void doJoinFromLeftThenUpdateFkThenRevertBack(final boolean leftJoin,
final boolean withHeaders) {
final Properties streamsConfig = getStreamsProperties(optimization, withHeaders);
final Topology topology = getTopology(streamsConfig, materialized ? "store" : null, leftJoin, rejoin, leftVersioned, rightVersioned);
try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(topology).withConfig(streamsConfig).build()) {
try (final TopologyTestDriver driver = new TopologyTestDriver(topology, streamsConfig)) {
final TestInputTopic<String, String> right = driver.createInputTopic(RIGHT_TABLE, new StringSerializer(), new StringSerializer());
final TestInputTopic<String, String> left = driver.createInputTopic(LEFT_TABLE, new StringSerializer(), new StringSerializer());
final TestOutputTopic<String, String> outputTopic = driver.createOutputTopic(OUTPUT, new StringDeserializer(), new StringDeserializer());
Expand Down Expand Up @@ -438,7 +437,7 @@ public void doJoinFromRightThenDeleteRightEntity(final boolean leftJoin,
final boolean transactional) {
final Properties streamsConfig = getStreamsProperties(optimization, withHeaders, transactional);
final Topology topology = getTopology(streamsConfig, materialized ? "store" : null, leftJoin, rejoin, leftVersioned, rightVersioned);
try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(topology).withConfig(streamsConfig).build()) {
try (final TopologyTestDriver driver = new TopologyTestDriver(topology, streamsConfig)) {
final TestInputTopic<String, String> right = driver.createInputTopic(RIGHT_TABLE, new StringSerializer(), new StringSerializer());
final TestInputTopic<String, String> left = driver.createInputTopic(LEFT_TABLE, new StringSerializer(), new StringSerializer());
final TestOutputTopic<String, String> outputTopic = driver.createOutputTopic(OUTPUT, new StringDeserializer(), new StringDeserializer());
Expand Down Expand Up @@ -559,7 +558,7 @@ public void shouldEmitTombstoneWhenDeletingNonJoiningRecords(final boolean leftJ
final boolean withHeaders) {
final Properties streamsConfig = getStreamsProperties(optimization, withHeaders);
final Topology topology = getTopology(streamsConfig, materialized ? "store" : null, leftJoin, rejoin, leftVersioned, rightVersioned);
try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(topology).withConfig(streamsConfig).build()) {
try (final TopologyTestDriver driver = new TopologyTestDriver(topology, streamsConfig)) {
final TestInputTopic<String, String> left = driver.createInputTopic(LEFT_TABLE, new StringSerializer(), new StringSerializer());
final TestOutputTopic<String, String> outputTopic = driver.createOutputTopic(OUTPUT, new StringDeserializer(), new StringDeserializer());
final KeyValueStore<String, ValueAndTimestamp<String>> store = driver.getTimestampedKeyValueStore("store");
Expand Down Expand Up @@ -626,7 +625,7 @@ public void shouldNotEmitTombstonesWhenDeletingNonExistingRecords(final boolean
final boolean withHeaders) {
final Properties streamsConfig = getStreamsProperties(optimization, withHeaders);
final Topology topology = getTopology(streamsConfig, materialized ? "store" : null, leftJoin, rejoin, leftVersioned, rightVersioned);
try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(topology).withConfig(streamsConfig).build()) {
try (final TopologyTestDriver driver = new TopologyTestDriver(topology, streamsConfig)) {
final TestInputTopic<String, String> left = driver.createInputTopic(LEFT_TABLE, new StringSerializer(), new StringSerializer());
final TestOutputTopic<String, String> outputTopic = driver.createOutputTopic(OUTPUT, new StringDeserializer(), new StringDeserializer());
final KeyValueStore<String, ValueAndTimestamp<String>> store = driver.getTimestampedKeyValueStore("store");
Expand Down Expand Up @@ -659,7 +658,7 @@ public void joinShouldProduceNullsWhenValueHasNonMatchingForeignKey(final boolea
final boolean withHeaders) {
final Properties streamsConfig = getStreamsProperties(optimization, withHeaders);
final Topology topology = getTopology(streamsConfig, materialized ? "store" : null, leftJoin, rejoin, leftVersioned, rightVersioned);
try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(topology).withConfig(streamsConfig).build()) {
try (final TopologyTestDriver driver = new TopologyTestDriver(topology, streamsConfig)) {
final TestInputTopic<String, String> right = driver.createInputTopic(RIGHT_TABLE, new StringSerializer(), new StringSerializer());
final TestInputTopic<String, String> left = driver.createInputTopic(LEFT_TABLE, new StringSerializer(), new StringSerializer());
final TestOutputTopic<String, String> outputTopic = driver.createOutputTopic(OUTPUT, new StringDeserializer(), new StringDeserializer());
Expand Down Expand Up @@ -767,7 +766,7 @@ public void shouldUnsubscribeOldForeignKeyIfLeftSideIsUpdated(final boolean left
final boolean withHeaders) {
final Properties streamsConfig = getStreamsProperties(optimization, withHeaders);
final Topology topology = getTopology(streamsConfig, materialized ? "store" : null, leftJoin, rejoin, leftVersioned, rightVersioned);
try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(topology).withConfig(streamsConfig).build()) {
try (final TopologyTestDriver driver = new TopologyTestDriver(topology, streamsConfig)) {
final TestInputTopic<String, String> right = driver.createInputTopic(RIGHT_TABLE, new StringSerializer(), new StringSerializer());
final TestInputTopic<String, String> left = driver.createInputTopic(LEFT_TABLE, new StringSerializer(), new StringSerializer());
final TestOutputTopic<String, String> outputTopic = driver.createOutputTopic(OUTPUT, new StringDeserializer(), new StringDeserializer());
Expand Down Expand Up @@ -853,7 +852,7 @@ public void shouldEmitRecordOnNullForeignKeyForLeftJoins(final String optimizati
final boolean withHeaders) {
final Properties streamsConfig = getStreamsProperties(optimization, withHeaders);
final Topology topology = getTopology(streamsConfig, materialized ? "store" : null, true, rejoin, leftVersioned, rightVersioned, value -> null);
try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(topology).withConfig(streamsConfig).build()) {
try (final TopologyTestDriver driver = new TopologyTestDriver(topology, streamsConfig)) {
final TestInputTopic<String, String> left = driver.createInputTopic(LEFT_TABLE, new StringSerializer(), new StringSerializer());
final TestOutputTopic<String, String> outputTopic = driver.createOutputTopic(OUTPUT, new StringDeserializer(), new StringDeserializer());
final KeyValueStore<String, ValueAndTimestamp<String>> store = driver.getTimestampedKeyValueStore("store");
Expand Down Expand Up @@ -892,7 +891,7 @@ public void shouldEmitRecordWhenOldAndNewFkDiffer(final String optimization,
};
final Properties streamsConfig = getStreamsProperties(optimization, withHeaders, transactional);
final Topology topology = getTopology(streamsConfig, materialized ? "store" : null, true, rejoin, leftVersioned, rightVersioned, foreignKeyExtractor);
try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(topology).withConfig(streamsConfig).build()) {
try (final TopologyTestDriver driver = new TopologyTestDriver(topology, streamsConfig)) {
final TestInputTopic<String, String> left = driver.createInputTopic(LEFT_TABLE, new StringSerializer(), new StringSerializer());
final TestOutputTopic<String, String> outputTopic = driver.createOutputTopic(OUTPUT, new StringDeserializer(), new StringDeserializer());
final KeyValueStore<String, ValueAndTimestamp<String>> store = driver.getTimestampedKeyValueStore("store");
Expand Down Expand Up @@ -1080,7 +1079,7 @@ public void shouldIgnoreOutOfOrderRecordsIffVersioned(final boolean leftJoin,
final boolean withHeaders) {
final Properties streamsConfig = getStreamsProperties(optimization, withHeaders);
final Topology topology = getTopology(streamsConfig, materialized ? "store" : null, leftJoin, rejoin, leftVersioned, rightVersioned);
try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(topology).withConfig(streamsConfig).build()) {
try (final TopologyTestDriver driver = new TopologyTestDriver(topology, streamsConfig)) {
final TestInputTopic<String, String> right = driver.createInputTopic(RIGHT_TABLE, new StringSerializer(), new StringSerializer());
final TestInputTopic<String, String> left = driver.createInputTopic(LEFT_TABLE, new StringSerializer(), new StringSerializer());
final TestOutputTopic<String, String> outputTopic = driver.createOutputTopic(OUTPUT, new StringDeserializer(), new StringDeserializer());
Expand Down
3 changes: 1 addition & 2 deletions ...e/kafka/streams/integration/KTableKTableForeignKeyJoinMaterializationIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,6 @@
import org.apache.kafka.streams.TestOutputTopic;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.TopologyTestDriver;
import org.apache.kafka.streams.TopologyTestDriverBuilder;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
Expand Down Expand Up @@ -75,7 +74,7 @@ public void before() {
public void shouldEmitTombstoneWhenDeletingNonJoiningRecords(final boolean materialized, final boolean queryable, final boolean withHeaders) {
StreamsTestUtils.maybeSetDslStoreFormatHeaders(streamsConfig, withHeaders);
final Topology topology = getTopology(streamsConfig, "store", materialized, queryable);
try (final TopologyTestDriver driver = new TopologyTestDriverBuilder(topology).withConfig(streamsConfig).build()) {
try (final TopologyTestDriver driver = new TopologyTestDriver(topology, streamsConfig)) {
final TestInputTopic<String, String> left = driver.createInputTopic(LEFT_TABLE, new StringSerializer(), new StringSerializer());
final TestOutputTopic<String, String> outputTopic = driver.createOutputTopic(OUTPUT, new StringDeserializer(), new StringDeserializer());
final KeyValueStore<String, String> store = driver.getKeyValueStore("store");
Expand Down
Loading
Loading