Skip to content
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

KAFKA-16448: Add ProcessingExceptionHandler in Streams configuration #16092

Merged
merged 17 commits into from
May 30, 2024
Merged

KAFKA-16448: Add ProcessingExceptionHandler in Streams configuration #16092

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
9cfac39
KAFKA-16448: Add ProcessingExceptionHandler interface and implementat…
loicgreffier May 27, 2024
b957861
KAFKA-16448: Add ProcessingExceptionHandler in Streams configuration
loicgreffier May 27, 2024
9eef86d
KAFKA-16448: Add ProcessingExceptionHandler interface and implementat…
loicgreffier May 27, 2024
40fdac5
KAFKA-16448: Update from parent branch
loicgreffier May 28, 2024
3cb797b
KAFKA-16448: Fix indentation
loicgreffier May 28, 2024
c1c0d5c
KAFKA-16448: Remove unecessary classes
loicgreffier May 28, 2024
bea1fb8
KAFKA-16448: add Streams configuration test for ProcessingExceptionHa…
sebastienviale May 29, 2024
b7058d2
KAFKA-16448: add Streams configuration test for ProcessingExceptionHa…
sebastienviale May 29, 2024
0a54bae
Merge branch 'KAFKA-16448-Add-Processing-Exception-Handler-StreamsCon…
sebastienviale May 29, 2024
014b799
Merge branch 'KAFKA-16448-Add-Processing-Exception-Handler-StreamsCon…
sebastienviale May 29, 2024
10e69a9
Merge branch 'KAFKA-16448-Add-Processing-Exception-Handler-StreamsCon…
sebastienviale May 29, 2024
3c7de0c
KAFKA-16448: add Streams configuration test for invalid ProcessingExc…
sebastienviale May 29, 2024
5e7a656
KAFKA-16448: add error message test in case of invalid ProcessingExce…
sebastienviale May 29, 2024
60bba3b
KAFKA-16448: Add ProcessingExceptionHandler in Streams configuration
loicgreffier May 27, 2024
25115ce
KAFKA-16448: Add ProcessingExceptionHandler in Streams configuration
loicgreffier May 29, 2024
4b56def
Merge branch 'KAFKA-16448-Add-Processing-Exception-Handler-StreamsCon…
sebastienviale May 29, 2024
7d67257
KAFKA-16448: remove unused imports
sebastienviale May 29, 2024
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
16 changes: 16 additions & 0 deletions streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,8 @@
import org.apache.kafka.streams.errors.DefaultProductionExceptionHandler;
import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
import org.apache.kafka.streams.errors.LogAndFailExceptionHandler;
import org.apache.kafka.streams.errors.LogAndFailProcessingExceptionHandler;
import org.apache.kafka.streams.errors.ProcessingExceptionHandler;
import org.apache.kafka.streams.errors.ProductionExceptionHandler;
import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.internals.StreamsConfigUtils;
Expand Down Expand Up @@ -554,6 +556,11 @@ public class StreamsConfig extends AbstractConfig {
public static final String DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG = "default.production.exception.handler";
private static final String DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_DOC = "Exception handling class that implements the <code>org.apache.kafka.streams.errors.ProductionExceptionHandler</code> interface.";

/** {@code processing.exception.handler} */
@SuppressWarnings("WeakerAccess")
public static final String PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG = "processing.exception.handler";
public static final String PROCESSING_EXCEPTION_HANDLER_CLASS_DOC = "Exception handling class that implements the <code>org.apache.kafka.streams.errors.ProcessingExceptionHandler</code> interface.";

/** {@code default.dsl.store} */
@Deprecated
@SuppressWarnings("WeakerAccess")
Expand Down Expand Up @@ -931,6 +938,11 @@ public class StreamsConfig extends AbstractConfig {
DefaultProductionExceptionHandler.class.getName(),
Importance.MEDIUM,
DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_DOC)
.define(PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG,
Type.CLASS,
LogAndFailProcessingExceptionHandler.class.getName(),
Importance.MEDIUM,
PROCESSING_EXCEPTION_HANDLER_CLASS_DOC)
Comment on lines +941 to +945
Copy link
Contributor

Choose a reason for hiding this comment

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

Could you please add unit tests in StreamsConfigTest?
I think you need the following unit tests:

  • One where config is not set to verify the default value
  • One where the config is set and you get the correct instance from processingExceptionHandler().

Copy link
Contributor

Choose a reason for hiding this comment

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

hi,
It is done !

.define(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG,
Type.CLASS,
FailOnInvalidTimestamp.class.getName(),
Expand Down Expand Up @@ -1925,6 +1937,10 @@ public ProductionExceptionHandler defaultProductionExceptionHandler() {
return getConfiguredInstance(DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG, ProductionExceptionHandler.class);
}

public ProcessingExceptionHandler processingExceptionHandler() {
return getConfiguredInstance(PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG, ProcessingExceptionHandler.class);
}

/**
* Override any client properties in the original configs with overrides
*
Expand Down
27 changes: 27 additions & 0 deletions streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1586,6 +1586,33 @@ public void shouldDisableMetricCollectionOnMainConsumerOnly() {
);
}

@Test
public void shouldGetDefaultValueProcessingExceptionHandler() {
final StreamsConfig streamsConfig = new StreamsConfig(props);

assertEquals("org.apache.kafka.streams.errors.LogAndFailProcessingExceptionHandler", streamsConfig.processingExceptionHandler().getClass().getName());
}

@Test
public void shouldOverrideDefaultProcessingExceptionHandler() {
props.put(StreamsConfig.PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG, "org.apache.kafka.streams.errors.LogAndContinueProcessingExceptionHandler");
final StreamsConfig streamsConfig = new StreamsConfig(props);

assertEquals("org.apache.kafka.streams.errors.LogAndContinueProcessingExceptionHandler", streamsConfig.processingExceptionHandler().getClass().getName());
}

@Test
public void testInvalidProcessingExceptionHandler() {
props.put(StreamsConfig.PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG, "org.apache.kafka.streams.errors.InvalidProcessingExceptionHandler");
final Exception exception = assertThrows(ConfigException.class, () -> new StreamsConfig(props));

assertThat(
exception.getMessage(),
containsString("Invalid value org.apache.kafka.streams.errors.InvalidProcessingExceptionHandler " +
"for configuration processing.exception.handler: Class org.apache.kafka.streams.errors.InvalidProcessingExceptionHandler could not be found.")
);
}

static class MisconfiguredSerde implements Serde<Object> {
@Override
public void configure(final Map<String, ?> configs, final boolean isKey) {
Expand Down