Skip to content

KAFKA-20329: Test headers dsl.store.format further (3/N) #21820

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
aliehsaeedii wants to merge 4 commits into
base: trunk
Choose a base branch
from
Open

KAFKA-20329: Test headers dsl.store.format further (3/N) #21820

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
fe1f31c
test headers mode
aliehsaeedii Mar 19, 2026
602ea31
address comment
aliehsaeedii Mar 26, 2026
f3f77a8
address comments 2
aliehsaeedii Apr 7, 2026
4e040bb
address comments
aliehsaeedii Apr 8, 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
35 changes: 26 additions & 9 deletions ...ts/src/test/java/org/apache/kafka/streams/integration/GlobalKTableEOSIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,9 +48,10 @@
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInfo;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;

import java.io.IOException;
import java.time.Duration;
Expand Down Expand Up @@ -136,8 +137,12 @@ public void after() throws Exception {
IntegrationTestUtils.purgeLocalStreamsState(streamsConfiguration);
}

@Test
public void shouldKStreamGlobalKTableLeftJoin() throws Exception {
@ParameterizedTest
@ValueSource(booleans = {false, true})
public void shouldKStreamGlobalKTableLeftJoin(final boolean withHeaders) throws Exception {
if (withHeaders) {
streamsConfiguration.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
}
Copy link
Copy Markdown
Contributor

@frankvicky frankvicky Mar 20, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we add a withHeader parameter to startStreams() for reducing the repeat code ?

final KStream<String, String> streamTableJoin = stream.leftJoin(globalTable, keyMapper, joiner);
streamTableJoin.foreach(foreachAction);
produceInitialGlobalTableValues();
Expand Down Expand Up @@ -207,8 +212,12 @@ public void shouldKStreamGlobalKTableLeftJoin() throws Exception {
);
}

@Test
public void shouldKStreamGlobalKTableJoin() throws Exception {
@ParameterizedTest
@ValueSource(booleans = {false, true})
public void shouldKStreamGlobalKTableJoin(final boolean withHeaders) throws Exception {
if (withHeaders) {
streamsConfiguration.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
}
final KStream<String, String> streamTableJoin = stream.join(globalTable, keyMapper, joiner);
streamTableJoin.foreach(foreachAction);
produceInitialGlobalTableValues();
Expand Down Expand Up @@ -277,8 +286,12 @@ public void shouldKStreamGlobalKTableJoin() throws Exception {
);
}

@Test
public void shouldRestoreTransactionalMessages() throws Exception {
@ParameterizedTest
@ValueSource(booleans = {false, true})
public void shouldRestoreTransactionalMessages(final boolean withHeaders) throws Exception {
if (withHeaders) {
streamsConfiguration.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
}
produceInitialGlobalTableValues();

startStreams();
Expand Down Expand Up @@ -309,8 +322,12 @@ public void shouldRestoreTransactionalMessages() throws Exception {
);
}

@Test
public void shouldNotRestoreAbortedMessages() throws Exception {
@ParameterizedTest
@ValueSource(booleans = {false, true})
public void shouldNotRestoreAbortedMessages(final boolean withHeaders) throws Exception {
if (withHeaders) {
streamsConfiguration.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
}
produceAbortedMessages();
produceInitialGlobalTableValues();
produceAbortedMessages();
Expand Down
89 changes: 68 additions & 21 deletions ...tests/src/test/java/org/apache/kafka/streams/integration/GlobalKTableIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,9 +54,10 @@
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInfo;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;

import java.io.IOException;
import java.time.Duration;
Expand Down Expand Up @@ -139,8 +140,12 @@ public void whenShuttingDown() throws Exception {
IntegrationTestUtils.purgeLocalStreamsState(streamsConfiguration);
}

@Test
public void shouldKStreamGlobalKTableLeftJoin() throws Exception {
@ParameterizedTest
@ValueSource(booleans = {false, true})
public void shouldKStreamGlobalKTableLeftJoin(final boolean withHeaders) throws Exception {
if (withHeaders) {
streamsConfiguration.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
}
final KStream<String, String> streamTableJoin = stream.leftJoin(globalTable, keyMapper, joiner);
streamTableJoin.process(supplier);
produceInitialGlobalTableValues();
Expand Down Expand Up @@ -224,8 +229,12 @@ public void shouldKStreamGlobalKTableLeftJoin() throws Exception {
"waiting for final values");
}

@Test
public void shouldKStreamGlobalKTableJoin() throws Exception {
@ParameterizedTest
@ValueSource(booleans = {false, true})
public void shouldKStreamGlobalKTableJoin(final boolean withHeaders) throws Exception {
if (withHeaders) {
streamsConfiguration.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
}
final KStream<String, String> streamTableJoin = stream.join(globalTable, keyMapper, joiner);
streamTableJoin.process(supplier);
produceInitialGlobalTableValues();
Expand Down Expand Up @@ -309,8 +318,12 @@ public void shouldKStreamGlobalKTableJoin() throws Exception {
"waiting for final values");
}

@Test
public void shouldRestoreGlobalInMemoryKTableOnRestart() throws Exception {
@ParameterizedTest
@ValueSource(booleans = {false, true})
public void shouldRestoreGlobalInMemoryKTableOnRestart(final boolean withHeaders) throws Exception {
if (withHeaders) {
streamsConfiguration.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
}
builder = new StreamsBuilder();
globalTable = builder.globalTable(
globalTableTopic,
Expand Down Expand Up @@ -340,8 +353,12 @@ public void shouldRestoreGlobalInMemoryKTableOnRestart() throws Exception {
assertThat(timestampedStore.approximateNumEntries(), equalTo(4L));
}

@Test
public void shouldGetToRunningWithOnlyGlobalTopology() throws Exception {
@ParameterizedTest
@ValueSource(booleans = {false, true})
public void shouldGetToRunningWithOnlyGlobalTopology(final boolean withHeaders) throws Exception {
if (withHeaders) {
streamsConfiguration.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
}
builder = new StreamsBuilder();
globalTable = builder.globalTable(
globalTableTopic,
Expand Down Expand Up @@ -375,31 +392,41 @@ public void process(final Record<Long, String> record) {
);
}

@Test
public void testProcessingExceptionHandlerContinueEnabledRestorationPhase() throws Exception {
@ParameterizedTest
@ValueSource(booleans = {false, true})
public void testProcessingExceptionHandlerContinueEnabledRestorationPhase(final boolean withHeaders) throws Exception {
createBuilderWithFailedProcessor();
// enable processing exception handler invoked config
TestGlobalProcessingExceptionHandler.shouldResume = true;
streamsConfiguration.put(StreamsConfig.PROCESSING_EXCEPTION_HANDLER_GLOBAL_ENABLED_CONFIG, true);
streamsConfiguration.put(StreamsConfig.PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG,
TestGlobalProcessingExceptionHandler.class);

if (withHeaders) {
streamsConfiguration.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
}

produceInitialGlobalTableValues();
startStreams();
waitForApplicationState(singletonList(kafkaStreams), State.RUNNING, Duration.ofSeconds(30));

assertTrue(TestGlobalProcessingExceptionHandler.handlerInvoked.get());
}

@Test
public void testProcessingExceptionHandlerFailEnabledRestorationPhase() throws Exception {
@ParameterizedTest
@ValueSource(booleans = {false, true})
public void testProcessingExceptionHandlerFailEnabledRestorationPhase(final boolean withHeaders) throws Exception {
createBuilderWithFailedProcessor();
// enable processing exception handler invoked config
TestGlobalProcessingExceptionHandler.shouldResume = false;
streamsConfiguration.put(StreamsConfig.PROCESSING_EXCEPTION_HANDLER_GLOBAL_ENABLED_CONFIG, true);
streamsConfiguration.put(StreamsConfig.PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG,
TestGlobalProcessingExceptionHandler.class);

if (withHeaders) {
streamsConfiguration.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
}

produceInitialGlobalTableValues();
assertThrows(StreamsException.class, () -> {
startStreams();
Expand All @@ -408,15 +435,20 @@ public void testProcessingExceptionHandlerFailEnabledRestorationPhase() throws E

}

@Test
public void testProcessingExceptionHandlerDisabledRestorationPhase() throws Exception {
@ParameterizedTest
@ValueSource(booleans = {false, true})
public void testProcessingExceptionHandlerDisabledRestorationPhase(final boolean withHeaders) throws Exception {
createBuilderWithFailedProcessor();
// disable processing exception handler invoked config
TestGlobalProcessingExceptionHandler.shouldResume = false;
streamsConfiguration.put(StreamsConfig.PROCESSING_EXCEPTION_HANDLER_GLOBAL_ENABLED_CONFIG, false);
streamsConfiguration.put(StreamsConfig.PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG,
TestGlobalProcessingExceptionHandler.class);

if (withHeaders) {
streamsConfiguration.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
}

produceInitialGlobalTableValues();
assertThrows(StreamsException.class, () -> {
startStreams();
Expand All @@ -425,15 +457,20 @@ public void testProcessingExceptionHandlerDisabledRestorationPhase() throws Exce

}

@Test
public void testProcessingExceptionHandlerContinueEnabledRunTimePhase() throws Exception {
@ParameterizedTest
@ValueSource(booleans = {false, true})
public void testProcessingExceptionHandlerContinueEnabledRunTimePhase(final boolean withHeaders) throws Exception {
createBuilderWithFailedProcessor();
// enable processing exception handler invoked config
TestGlobalProcessingExceptionHandler.shouldResume = true;
streamsConfiguration.put(StreamsConfig.PROCESSING_EXCEPTION_HANDLER_GLOBAL_ENABLED_CONFIG, true);
streamsConfiguration.put(StreamsConfig.PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG,
TestGlobalProcessingExceptionHandler.class);

if (withHeaders) {
streamsConfiguration.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
}

startStreams();
waitForApplicationState(singletonList(kafkaStreams), State.RUNNING, Duration.ofSeconds(30));
produceInitialGlobalTableValues();
Expand All @@ -445,29 +482,39 @@ public void testProcessingExceptionHandlerContinueEnabledRunTimePhase() throws E
);
}

@Test
public void testProcessingExceptionHandlerFailEnabledRunTimePhase() throws Exception {
@ParameterizedTest
@ValueSource(booleans = {false, true})
public void testProcessingExceptionHandlerFailEnabledRunTimePhase(final boolean withHeaders) throws Exception {
createBuilderWithFailedProcessor();
// enable processing exception handler invoked config
streamsConfiguration.put(StreamsConfig.PROCESSING_EXCEPTION_HANDLER_GLOBAL_ENABLED_CONFIG, true);
streamsConfiguration.put(StreamsConfig.PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG,
TestGlobalProcessingExceptionHandler.class);

if (withHeaders) {
streamsConfiguration.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
}

startStreams();
waitForApplicationState(singletonList(kafkaStreams), State.RUNNING, Duration.ofSeconds(30));
produceInitialGlobalTableValues();
waitForApplicationState(singletonList(kafkaStreams), State.ERROR, Duration.ofSeconds(30));
assertTrue(TestGlobalProcessingExceptionHandler.handlerInvoked.get());
}

@Test
public void testProcessingExceptionHandlerDisabledRunTimePhase() throws Exception {
@ParameterizedTest
@ValueSource(booleans = {false, true})
public void testProcessingExceptionHandlerDisabledRunTimePhase(final boolean withHeaders) throws Exception {
createBuilderWithFailedProcessor();
// disable processing exception handler invoked config
streamsConfiguration.put(StreamsConfig.PROCESSING_EXCEPTION_HANDLER_GLOBAL_ENABLED_CONFIG, false);
streamsConfiguration.put(StreamsConfig.PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG,
TestGlobalProcessingExceptionHandler.class);

if (withHeaders) {
streamsConfiguration.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
}

startStreams();
waitForApplicationState(singletonList(kafkaStreams), State.RUNNING, Duration.ofSeconds(30));
produceInitialGlobalTableValues();
Expand Down
25 changes: 12 additions & 13 deletions ...ests/src/test/java/org/apache/kafka/streams/integration/InternalTopicIntegrationTest.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.common.serialization.StringSerializer;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.server.util.MockTime;
import org.apache.kafka.streams.GroupProtocol;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
Expand Down Expand Up @@ -149,10 +148,10 @@ private Admin createAdminClient() {
return Admin.create(adminClientConfig);
}

private void configureStreams(final boolean streamsProtocolEnabled, final String appID) {
private void configureStreams(final boolean withHeaders, final String appID) {
Copy link
Copy Markdown
Member

@mjsax mjsax Apr 6, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why do we remove streamsProtocolEnabled ?

Copy link
Copy Markdown
Contributor Author

@aliehsaeedii aliehsaeedii Apr 7, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

:(

Copy link
Copy Markdown
Contributor Author

@aliehsaeedii aliehsaeedii Apr 7, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

:O

streamsProp.put(StreamsConfig.APPLICATION_ID_CONFIG, appID);
if (streamsProtocolEnabled) {
streamsProp.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name().toLowerCase(Locale.getDefault()));
if (withHeaders) {
streamsProp.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
}
}

Expand All @@ -162,9 +161,9 @@ private void configureStreams(final boolean streamsProtocolEnabled, final String
*/
@ParameterizedTest
@ValueSource(booleans = {true, false})
public void shouldGetToRunningWithWindowedTableInFKJ(final boolean streamsProtocolEnabled) throws Exception {
final String appID = APP_ID + "-windowed-FKJ-" + streamsProtocolEnabled;
configureStreams(streamsProtocolEnabled, appID);
public void shouldGetToRunningWithWindowedTableInFKJ(final boolean withHeaders) throws Exception {
final String appID = APP_ID + "-windowed-FKJ-" + withHeaders;
configureStreams(withHeaders, appID);

final StreamsBuilder streamsBuilder = new StreamsBuilder();
final KStream<String, String> inputTopic = streamsBuilder.stream(DEFAULT_INPUT_TOPIC);
Expand Down Expand Up @@ -192,9 +191,9 @@ public void shouldGetToRunningWithWindowedTableInFKJ(final boolean streamsProtoc

@ParameterizedTest
@ValueSource(booleans = {true, false})
public void shouldCompactTopicsForKeyValueStoreChangelogs(final boolean streamsProtocolEnabled) throws Exception {
final String appID = APP_ID + "-compact-" + streamsProtocolEnabled;
configureStreams(streamsProtocolEnabled, appID);
public void shouldCompactTopicsForKeyValueStoreChangelogs(final boolean withHeaders) throws Exception {
final String appID = APP_ID + "-compact-" + withHeaders;
configureStreams(withHeaders, appID);

//
// Step 1: Configure and start a simple word count topology
Expand Down Expand Up @@ -230,9 +229,9 @@ public void shouldCompactTopicsForKeyValueStoreChangelogs(final boolean streamsP

@ParameterizedTest
@ValueSource(booleans = {true, false})
public void shouldCompactAndDeleteTopicsForWindowStoreChangelogs(final boolean streamsProtocolEnabled) throws Exception {
final String appID = APP_ID + "-compact-delete-" + streamsProtocolEnabled;
configureStreams(streamsProtocolEnabled, appID);
public void shouldCompactAndDeleteTopicsForWindowStoreChangelogs(final boolean withHeaders) throws Exception {
final String appID = APP_ID + "-compact-delete-" + withHeaders;
configureStreams(withHeaders, appID);

//
// Step 1: Configure and start a simple word count topology
Expand Down
20 changes: 9 additions & 11 deletions ...tion-tests/src/test/java/org/apache/kafka/streams/integration/MetricsIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,6 @@
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.streams.GroupProtocol;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KafkaStreams.State;
import org.apache.kafka.streams.KeyValue;
Expand Down Expand Up @@ -59,7 +58,6 @@
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Locale;
import java.util.Properties;
import java.util.stream.Collectors;

Expand Down Expand Up @@ -344,9 +342,9 @@ private void closeApplication() throws Exception {

@ParameterizedTest
@ValueSource(booleans = {true, false})
public void shouldAddMetricsOnAllLevels(final boolean streamsProtocolEnabled) throws Exception {
if (streamsProtocolEnabled) {
Copy link
Copy Markdown
Member

@mjsax mjsax Apr 6, 2026 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same question? Also below.

streamsConfiguration.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name().toLowerCase(Locale.getDefault()));
public void shouldAddMetricsOnAllLevels(final boolean withHeaders) throws Exception {
if (withHeaders) {
streamsConfiguration.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
}

builder.stream(STREAM_INPUT, Consumed.with(Serdes.Integer(), Serdes.String()))
Expand Down Expand Up @@ -383,9 +381,9 @@ public void shouldAddMetricsOnAllLevels(final boolean streamsProtocolEnabled) th

@ParameterizedTest
@ValueSource(booleans = {true, false})
public void shouldAddMetricsForWindowStoreAndSuppressionBuffer(final boolean streamsProtocolEnabled) throws Exception {
if (streamsProtocolEnabled) {
streamsConfiguration.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name().toLowerCase(Locale.getDefault()));
public void shouldAddMetricsForWindowStoreAndSuppressionBuffer(final boolean withHeaders) throws Exception {
if (withHeaders) {
streamsConfiguration.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
}

final Duration windowSize = Duration.ofMillis(50);
Expand Down Expand Up @@ -416,9 +414,9 @@ public void shouldAddMetricsForWindowStoreAndSuppressionBuffer(final boolean str

@ParameterizedTest
@ValueSource(booleans = {true, false})
public void shouldAddMetricsForSessionStore(final boolean streamsProtocolEnabled) throws Exception {
if (streamsProtocolEnabled) {
streamsConfiguration.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name().toLowerCase(Locale.getDefault()));
public void shouldAddMetricsForSessionStore(final boolean withHeaders) throws Exception {
if (withHeaders) {
streamsConfiguration.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
}

final Duration inactivityGap = Duration.ofMillis(50);
Expand Down
Loading
Loading