You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "muralibasani (via GitHub)" <gi...@apache.org> on 2024/08/26 17:35:59 UTC
[PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
muralibasani opened a new pull request, #17005:
URL: https://github.com/apache/kafka/pull/17005
Resolves : https://issues.apache.org/jira/browse/KAFKA-16863
- Removing `default.` prefix for exception handler config
- Related kip removing `default.` prefix for exception handler config
*More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.*
*Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.*
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1734172808
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1907,13 +1937,38 @@ public TimestampExtractor defaultTimestampExtractor() {
return getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
}
+ public DeserializationExceptionHandler deserializationExceptionHandler() {
+ if (getClass(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null &&
Review Comment:
thanks for the tip.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1735363445
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -531,15 +531,33 @@ public class StreamsConfig extends AbstractConfig {
public static final String DEFAULT_CLIENT_SUPPLIER_CONFIG = "default.client.supplier";
public static final String DEFAULT_CLIENT_SUPPLIER_DOC = "Client supplier class that implements the <code>org.apache.kafka.streams.KafkaClientSupplier</code> interface.";
- /** {@code default.deserialization.exception.handler} */
+ /**
+ * {@code default.deserialization.exception.handler}
+ * @deprecated since 4.0; use {@link #DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG} instead
+ */
@SuppressWarnings("WeakerAccess")
+ @Deprecated
public static final String DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG = "default.deserialization.exception.handler";
+ @Deprecated
public static final String DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC = "Exception handling class that implements the <code>org.apache.kafka.streams.errors.DeserializationExceptionHandler</code> interface.";
- /** {@code default.production.exception.handler} */
+ /** {@code deserialization.exception.handler} */
@SuppressWarnings("WeakerAccess")
+ public static final String DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG = "deserialization.exception.handler";
+ static final String DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC = "Exception handling class that implements the <code>org.apache.kafka.streams.errors.DeserializationExceptionHandler</code> interface.";
+
+ /**
+ * {@code default.production.exception.handler}
+ * @deprecated since 4.0. Use {@link #PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG} instead
Review Comment:
```suggestion
* @deprecated since 4.0; use {@link #PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG} instead
```
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1910,11 +1938,45 @@ public TimestampExtractor defaultTimestampExtractor() {
return getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
}
+ public DeserializationExceptionHandler deserializationExceptionHandler() {
+ if (originals().containsKey(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) &&
+ originals().containsKey(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG)) {
+ log.warn("Both the deprecated and new config for deserialization exception handler are configured. " +
+ "The deprecated one will be ignored.");
+ }
+ if (originals().containsKey(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG)) {
+ return getConfiguredInstance(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
+ } else {
+ return defaultDeserializationExceptionHandler();
+ }
+ }
+
+ /**
+ * @deprecated as of kafka 4.0; use {@link #deserializationExceptionHandler()} instead
Review Comment:
```suggestion
* @deprecated since 4.0; use {@link #deserializationExceptionHandler()} instead
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1734129356
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -889,6 +901,11 @@ public class StreamsConfig extends AbstractConfig {
LogAndFailExceptionHandler.class.getName(),
Importance.MEDIUM,
DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC)
+ .define(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
+ Type.CLASS,
+ LogAndFailExceptionHandler.class.getName(),
Review Comment:
Currently we have set default values for both old and new config which is LogAndFailExceptionHandler.class.
With this situation, even if user configures a different handler to old config, LogAndFailExceptionHandler will be picked up.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1732517413
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java:
##########
@@ -1158,6 +1159,75 @@ public void shouldSkipOnDeserializationErrorsWhenReprocessing() {
assertEquals(0, stateRestoreCallback.restored.size());
}
+ @SuppressWarnings("deprecation")
+ @Test
+ public void verifyExceptionHandlerAcceptNewConfigWhenBothArePresent() {
Review Comment:
Moved test to StreamsConfigTest, agree.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1734159025
##########
streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java:
##########
@@ -1622,6 +1624,30 @@ public void testInvalidProcessingExceptionHandler() {
);
}
+ @Test
+ public void testDeserializationExceptionHandlerWhenOnlyNewConfigIsSet() {
+ props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndContinueExceptionHandler.class, streamsConfig.deserializationExceptionHandler().getClass());
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void testDeserializationExceptionHandlerWhenBothConfigsAreSet() {
+ props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+ props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndFailExceptionHandler.class);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndContinueExceptionHandler.class, streamsConfig.deserializationExceptionHandler().getClass());
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void testDeserializationExceptionHandlerWhenOnlyOldConfigIsSet() {
+ props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndFailExceptionHandler.class, streamsConfig.deserializationExceptionHandler().getClass());
Review Comment:
Indeed, I knew it, but was waiting for my other comment. After using originals, yes, it is fixed. Updated. Thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1742817450
##########
docs/streams/upgrade-guide.html:
##########
@@ -135,6 +135,12 @@ <h3 class="anchor-heading"><a id="streams_notable_changes" class="anchor-link"><
<h3><a id="streams_api_changes_400" href="#streams_api_changes_400">Streams API changes in 4.0.0</a></h3>
+ <p>
+ In this release two configs default.deserialization.exception.handler and default.production.exception.handler are deprecated, as they don't have any overwrites, which is described in
+ <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-1056%3A+Remove+%60default.%60+prefix+for+exception+handler+StreamsConfig">KIP-1056</a>
+ You can refer to new configs via <code>StreamsConfig#DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG</code> and <code>StreamsConfig#PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG</code>.
Review Comment:
I think it's better to use config name `<code>deserialization.exception.handler</code>` directly, and not refer to `StreamsConfig` -- in the end, the config name changed, what is the central thing (eg, for config files) -- it's only a "side effect" that we also add new variables to `StreamsConfig`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on PR #17005:
URL: https://github.com/apache/kafka/pull/17005#issuecomment-2336431350
Pushed that change. Will get site-docs running locally from next time. Thought this would be a smaller one.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1734147802
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1907,13 +1937,38 @@ public TimestampExtractor defaultTimestampExtractor() {
return getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
}
+ public DeserializationExceptionHandler deserializationExceptionHandler() {
+ if (getClass(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null &&
+ getClass(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null) {
+ log.warn("Both the deprecated and new config for deserialization exception handler are configured !!");
+ }
+ if (getClass(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null) {
+ return getConfiguredInstance(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
+ } else {
+ return defaultDeserializationExceptionHandler();
+ }
+ }
+
+ /**
+ * @deprecated as of kafka 4.0. Use deserializationExceptionHandler() instead
+ * @return DeserializationExceptionHandler
+ */
+ @Deprecated
@SuppressWarnings("WeakerAccess")
public DeserializationExceptionHandler defaultDeserializationExceptionHandler() {
return getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
}
+ public ProductionExceptionHandler productionExceptionHandler() {
+ if (getClass(PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG) != null) {
Review Comment:
Ah yes. Updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1732520717
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilderTest.java:
##########
@@ -1066,6 +1067,41 @@ public void shouldOverrideGlobalStreamsConfigWhenGivenNamedTopologyProps() {
assertThat(topologyBuilder.topologyConfigs().parseStoreType(), equalTo(Materialized.StoreType.IN_MEMORY));
}
+ @SuppressWarnings("deprecation")
+ @Test
+ public void exceptionHandlerShouldAcceptNewConfig() {
+ final Properties topologyOverrides = new Properties();
+ topologyOverrides.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndFailExceptionHandler.class);
+ topologyOverrides.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+
+ final StreamsConfig config = new StreamsConfig(StreamsTestUtils.getStreamsConfig());
+ final InternalTopologyBuilder topologyBuilder = new InternalTopologyBuilder(
+ new TopologyConfig(
+ "my-topology",
+ config,
+ topologyOverrides)
+ );
+
+ assertThat(topologyBuilder.topologyConfigs().getTaskConfig().deserializationExceptionHandler.getClass(), equalTo(LogAndContinueExceptionHandler.class));
+ }
+
+ @Test
+ public void exceptionHandlerShouldAcceptNewConfigNoOtherDeprecatedConfigPresent() {
Review Comment:
Updated existing shouldNotOverrideGlobalStreamsConfigWhenGivenUnnamedTopologyProps
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1732507720
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -889,6 +901,11 @@ public class StreamsConfig extends AbstractConfig {
LogAndFailExceptionHandler.class.getName(),
Importance.MEDIUM,
DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC)
+ .define(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
+ Type.CLASS,
+ LogAndFailExceptionHandler.class.getName(),
Review Comment:
@mjsax if we configure a default like this, always new one will be picked up. What do you suggest ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1742819290
##########
docs/streams/developer-guide/config-streams.html:
##########
@@ -83,6 +83,7 @@
<li><a class="reference internal" href="#probing-rebalance-interval-ms" id="id30">probing.rebalance.interval.ms</a></li>
<li><a class="reference internal" href="#processing-exception-handler" id="id41">processing.exception.handler</a></li>
<li><a class="reference internal" href="#processing-guarantee" id="id25">processing.guarantee</a></li>
+ <li><a class="reference internal" href="#production-exception-handler" id="id43">production.exception.handler</a></li>
Review Comment:
This is still adding a new link `<a ... href="#production-exception-handler" ...>` which is dangling...
I think we should update the HTML here to use `production-exception-handler` as new anchor point: https://kafka.apache.org/38/documentation/streams/developer-guide/config-streams.html#default-production-exception-handler
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1746022773
##########
docs/streams/developer-guide/config-streams.html:
##########
@@ -70,11 +70,10 @@
<li><a class="reference internal" href="#optional-configuration-parameters" id="id6">Optional configuration parameters</a>
<ul>
<li><a class="reference internal" href="#acceptable-recovery-lag" id="id27">acceptable.recovery.lag</a></li>
- <li><a class="reference internal" href="#default-deserialization-exception-handler" id="id7">default.deserialization.exception.handler</a></li>
Review Comment:
it was a mistake, sorted based on the href link which was incorrect. fixed it nw.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1734124226
##########
streams/src/main/java/org/apache/kafka/streams/TopologyConfig.java:
##########
@@ -93,6 +95,11 @@ public class TopologyConfig extends AbstractConfig {
null,
Importance.MEDIUM,
DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC)
+ .define(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
Review Comment:
Missed it. Updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1735367199
##########
streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java:
##########
@@ -1622,6 +1626,72 @@ public void testInvalidProcessingExceptionHandler() {
);
}
+ @Test
+ public void shouldSetAndGetDeserializationExceptionHandlerWhenOnlyNewConfigIsSet() {
+ props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndContinueExceptionHandler.class, streamsConfig.deserializationExceptionHandler().getClass());
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void shouldUseNewDeserializationExceptionHandlerWhenBothConfigsAreSet() {
+ props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+ props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndFailExceptionHandler.class);
+
+ try (LogCaptureAppender streamsConfigLogs = LogCaptureAppender.createAndRegister(StreamsConfig.class)) {
+ streamsConfigLogs.setClassLogger(StreamsConfig.class, Level.WARN);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndContinueExceptionHandler.class, streamsConfig.deserializationExceptionHandler().getClass());
+
+ final long warningMessageWhenBothConfigsAreSet = streamsConfigLogs.getMessages().stream()
+ .filter(m -> m.contains("Both the deprecated and new config for deserialization exception handler are configured."))
+ .count();
+ assertEquals(1, warningMessageWhenBothConfigsAreSet);
+ }
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void shouldUseOldDeserializationExceptionHandlerWhenOnlyOldConfigIsSet() {
+ props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndContinueExceptionHandler.class, streamsConfig.deserializationExceptionHandler().getClass());
+ }
+
+ @Test
+ public void shouldSetAndGetProductionExceptionHandlerWhenOnlyNewConfigIsSet() {
+ props.put(StreamsConfig.PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG, DefaultProductionExceptionHandler.class);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(DefaultProductionExceptionHandler.class, streamsConfig.productionExceptionHandler().getClass());
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void shouldUseNewProductionExceptionHandlerWhenBothConfigsAreSet() {
+ props.put(StreamsConfig.PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG, DefaultProductionExceptionHandler.class);
+ props.put(StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG, ProductionExceptionHandler.class);
Review Comment:
Passing an interface does not seems to be right? For this case, the config is ignored as the new config is used, but otherwise, when we try to create an object we would crash.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1735795478
##########
streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java:
##########
@@ -1622,6 +1626,72 @@ public void testInvalidProcessingExceptionHandler() {
);
}
+ @Test
+ public void shouldSetAndGetDeserializationExceptionHandlerWhenOnlyNewConfigIsSet() {
+ props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndContinueExceptionHandler.class, streamsConfig.deserializationExceptionHandler().getClass());
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void shouldUseNewDeserializationExceptionHandlerWhenBothConfigsAreSet() {
+ props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+ props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndFailExceptionHandler.class);
+
+ try (LogCaptureAppender streamsConfigLogs = LogCaptureAppender.createAndRegister(StreamsConfig.class)) {
+ streamsConfigLogs.setClassLogger(StreamsConfig.class, Level.WARN);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndContinueExceptionHandler.class, streamsConfig.deserializationExceptionHandler().getClass());
+
+ final long warningMessageWhenBothConfigsAreSet = streamsConfigLogs.getMessages().stream()
+ .filter(m -> m.contains("Both the deprecated and new config for deserialization exception handler are configured."))
+ .count();
+ assertEquals(1, warningMessageWhenBothConfigsAreSet);
+ }
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void shouldUseOldDeserializationExceptionHandlerWhenOnlyOldConfigIsSet() {
+ props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndContinueExceptionHandler.class, streamsConfig.deserializationExceptionHandler().getClass());
+ }
+
+ @Test
+ public void shouldSetAndGetProductionExceptionHandlerWhenOnlyNewConfigIsSet() {
+ props.put(StreamsConfig.PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG, DefaultProductionExceptionHandler.class);
Review Comment:
Thought about it, but missed. Now updated with ProductionExceptionHandlerMock, and also had to add a no args constructor.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1731648217
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -555,12 +555,24 @@ public class StreamsConfig extends AbstractConfig {
/** {@code default.deserialization.exception.handler} */
Review Comment:
We should update the JavaDocs, too. Using `@deprecated` annotation and pointing to the new config to be used.
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1907,16 +1929,42 @@ public TimestampExtractor defaultTimestampExtractor() {
return getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
}
+ public DeserializationExceptionHandler getDeserializationExceptionHandler() {
+ if (getClass(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null) {
+ return deserializationExceptionHandler();
+ } else {
+ return defaultDeserializationExceptionHandler();
+ }
+ }
+
@SuppressWarnings("WeakerAccess")
- public DeserializationExceptionHandler defaultDeserializationExceptionHandler() {
+ private DeserializationExceptionHandler defaultDeserializationExceptionHandler() {
return getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
}
@SuppressWarnings("WeakerAccess")
- public ProductionExceptionHandler defaultProductionExceptionHandler() {
+ public DeserializationExceptionHandler deserializationExceptionHandler() {
+ return getConfiguredInstance(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
+ }
+
+ public ProductionExceptionHandler getProductionExceptionHandler() {
Review Comment:
Some comments as above
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java:
##########
@@ -2552,7 +2552,8 @@ public Set<TopicPartition> partitions() {
}
@ParameterizedTest
- @MethodSource("data")
+ @MethodSource("data")
+ @SuppressWarnings("deprecation")
Review Comment:
This test verifies that the handler is doing the right thing. Thus, we should update the test to use the new config and not add this annotation
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -555,12 +555,24 @@ public class StreamsConfig extends AbstractConfig {
/** {@code default.deserialization.exception.handler} */
@SuppressWarnings("WeakerAccess")
+ @Deprecated
public static final String DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG = "default.deserialization.exception.handler";
+ @Deprecated
public static final String DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC = "Exception handling class that implements the <code>org.apache.kafka.streams.errors.DeserializationExceptionHandler</code> interface.";
+ /** {@code deserialization.exception.handler} */
+ @SuppressWarnings("WeakerAccess")
+ public static final String DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG = "deserialization.exception.handler";
+
Review Comment:
nit: no empty line
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -889,6 +901,11 @@ public class StreamsConfig extends AbstractConfig {
LogAndFailExceptionHandler.class.getName(),
Importance.MEDIUM,
DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC)
+ .define(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
Review Comment:
Please insert at the right place (we keep it ordered alphabetically)
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -555,12 +555,24 @@ public class StreamsConfig extends AbstractConfig {
/** {@code default.deserialization.exception.handler} */
@SuppressWarnings("WeakerAccess")
+ @Deprecated
public static final String DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG = "default.deserialization.exception.handler";
+ @Deprecated
public static final String DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC = "Exception handling class that implements the <code>org.apache.kafka.streams.errors.DeserializationExceptionHandler</code> interface.";
+ /** {@code deserialization.exception.handler} */
+ @SuppressWarnings("WeakerAccess")
+ public static final String DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG = "deserialization.exception.handler";
+
+ protected static final String DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC = "Exception handling class that implements the <code>org.apache.kafka.streams.errors.DeserializationExceptionHandler</code> interface.";
/** {@code default.production.exception.handler} */
@SuppressWarnings("WeakerAccess")
+ @Deprecated
public static final String DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG = "default.production.exception.handler";
+
+ /** {@code production.exception.handler} */
+ @SuppressWarnings("WeakerAccess")
+ public static final String PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG = "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.";
Review Comment:
nit: rename variable, dropping `DEFAULT_` to align to config name.
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1907,16 +1929,42 @@ public TimestampExtractor defaultTimestampExtractor() {
return getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
}
+ public DeserializationExceptionHandler getDeserializationExceptionHandler() {
+ if (getClass(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null) {
+ return deserializationExceptionHandler();
+ } else {
+ return defaultDeserializationExceptionHandler();
+ }
+ }
+
@SuppressWarnings("WeakerAccess")
- public DeserializationExceptionHandler defaultDeserializationExceptionHandler() {
+ private DeserializationExceptionHandler defaultDeserializationExceptionHandler() {
Review Comment:
This is public API, and we cannot just make it private -- we can only deprecate it -- as above, must be covered by the KIP
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -555,12 +555,24 @@ public class StreamsConfig extends AbstractConfig {
/** {@code default.deserialization.exception.handler} */
@SuppressWarnings("WeakerAccess")
+ @Deprecated
public static final String DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG = "default.deserialization.exception.handler";
+ @Deprecated
public static final String DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC = "Exception handling class that implements the <code>org.apache.kafka.streams.errors.DeserializationExceptionHandler</code> interface.";
+ /** {@code deserialization.exception.handler} */
+ @SuppressWarnings("WeakerAccess")
+ public static final String DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG = "deserialization.exception.handler";
+
+ protected static final String DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC = "Exception handling class that implements the <code>org.apache.kafka.streams.errors.DeserializationExceptionHandler</code> interface.";
Review Comment:
Why `protected` -- should be `private`?
nit: missing empty line below.
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -555,12 +555,24 @@ public class StreamsConfig extends AbstractConfig {
/** {@code default.deserialization.exception.handler} */
@SuppressWarnings("WeakerAccess")
+ @Deprecated
public static final String DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG = "default.deserialization.exception.handler";
+ @Deprecated
public static final String DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC = "Exception handling class that implements the <code>org.apache.kafka.streams.errors.DeserializationExceptionHandler</code> interface.";
+ /** {@code deserialization.exception.handler} */
+ @SuppressWarnings("WeakerAccess")
+ public static final String DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG = "deserialization.exception.handler";
+
+ protected static final String DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC = "Exception handling class that implements the <code>org.apache.kafka.streams.errors.DeserializationExceptionHandler</code> interface.";
/** {@code default.production.exception.handler} */
Review Comment:
Update JavaDocs?
##########
streams/src/main/java/org/apache/kafka/streams/TopologyConfig.java:
##########
@@ -93,6 +95,11 @@ public class TopologyConfig extends AbstractConfig {
null,
Importance.MEDIUM,
DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC)
+ .define(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
Review Comment:
please insert at right place (alphabetically)
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -919,6 +936,11 @@ public class StreamsConfig extends AbstractConfig {
DefaultProductionExceptionHandler.class.getName(),
Importance.MEDIUM,
DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_DOC)
+ .define(PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG,
Review Comment:
Please insert at the right place (we keep it ordered alphabetically)
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordDeserializer.java:
##########
@@ -113,6 +115,7 @@ public static void handleDeserializationFailure(final DeserializationExceptionHa
throw new StreamsException("Deserialization exception handler is set to fail upon" +
" a deserialization error. If you would rather have the streaming pipeline" +
" continue after a deserialization error, please set the " +
+ DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG + " or the deprecated exception handler " +
Review Comment:
We should just point to the new one and not refer to the deprecated one at all. (For this case, we also don't need the `@SuppressWarnings` annotation
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1907,16 +1929,42 @@ public TimestampExtractor defaultTimestampExtractor() {
return getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
}
+ public DeserializationExceptionHandler getDeserializationExceptionHandler() {
Review Comment:
```suggestion
public DeserializationExceptionHandler deserializationExceptionHandler() {
```
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1907,16 +1929,42 @@ public TimestampExtractor defaultTimestampExtractor() {
return getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
}
+ public DeserializationExceptionHandler getDeserializationExceptionHandler() {
+ if (getClass(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null) {
+ return deserializationExceptionHandler();
+ } else {
+ return defaultDeserializationExceptionHandler();
+ }
+ }
+
@SuppressWarnings("WeakerAccess")
- public DeserializationExceptionHandler defaultDeserializationExceptionHandler() {
+ private DeserializationExceptionHandler defaultDeserializationExceptionHandler() {
return getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
}
@SuppressWarnings("WeakerAccess")
- public ProductionExceptionHandler defaultProductionExceptionHandler() {
+ public DeserializationExceptionHandler deserializationExceptionHandler() {
Review Comment:
Seems to be redundant to `[get]DeserializationExceptionHandler` above?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java:
##########
@@ -1158,6 +1159,75 @@ public void shouldSkipOnDeserializationErrorsWhenReprocessing() {
assertEquals(0, stateRestoreCallback.restored.size());
}
+ @SuppressWarnings("deprecation")
+ @Test
+ public void verifyExceptionHandlerAcceptNewConfigWhenBothArePresent() {
Review Comment:
This test raised the question if we should log a WARN if both are set?
But why do we test this on `GlobalStateManagerImplTest`? Seems this should go into `StreamsConfigTest`?
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1907,16 +1929,42 @@ public TimestampExtractor defaultTimestampExtractor() {
return getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
}
+ public DeserializationExceptionHandler getDeserializationExceptionHandler() {
Review Comment:
We should not use `get` prefix.
Also, this is public API and the KIP must cover it -- can you update the KIP accordingly, and send an update to the VOTE thread?
Or: we don't add it as `public` -- I personally always found that having these method being `public` is a leaky abstraction.... But there was some disagreement about it: https://github.com/apache/kafka/pull/14548
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilderTest.java:
##########
@@ -1066,6 +1067,41 @@ public void shouldOverrideGlobalStreamsConfigWhenGivenNamedTopologyProps() {
assertThat(topologyBuilder.topologyConfigs().parseStoreType(), equalTo(Materialized.StoreType.IN_MEMORY));
}
+ @SuppressWarnings("deprecation")
+ @Test
+ public void exceptionHandlerShouldAcceptNewConfig() {
+ final Properties topologyOverrides = new Properties();
+ topologyOverrides.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndFailExceptionHandler.class);
+ topologyOverrides.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+
+ final StreamsConfig config = new StreamsConfig(StreamsTestUtils.getStreamsConfig());
+ final InternalTopologyBuilder topologyBuilder = new InternalTopologyBuilder(
+ new TopologyConfig(
+ "my-topology",
+ config,
+ topologyOverrides)
+ );
+
+ assertThat(topologyBuilder.topologyConfigs().getTaskConfig().deserializationExceptionHandler.getClass(), equalTo(LogAndContinueExceptionHandler.class));
+ }
+
+ @Test
+ public void exceptionHandlerShouldAcceptNewConfigNoOtherDeprecatedConfigPresent() {
Review Comment:
Instead of adding a new test, should we just update an existing test which uses the old config now, and let is use the new config instead (maybe `shouldOverrideGlobalStreamsConfigWhenGivenNamedTopologyProps`) ?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java:
##########
@@ -259,6 +259,11 @@ public void setDeserializationExceptionHandler(final DeserializationExceptionHan
this.deserializationExceptionHandler = deserializationExceptionHandler;
}
+ //Visible for testing
+ public DeserializationExceptionHandler getDeserializationExceptionHandler() {
Review Comment:
```suggestion
public DeserializationExceptionHandler deserializationExceptionHandler() {
```
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilderTest.java:
##########
@@ -1066,6 +1067,41 @@ public void shouldOverrideGlobalStreamsConfigWhenGivenNamedTopologyProps() {
assertThat(topologyBuilder.topologyConfigs().parseStoreType(), equalTo(Materialized.StoreType.IN_MEMORY));
}
+ @SuppressWarnings("deprecation")
+ @Test
+ public void exceptionHandlerShouldAcceptNewConfig() {
Review Comment:
```suggestion
public void newDeserializationExceptionHandlerConfigShouldOverwriteOldOne() {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1744338416
##########
docs/streams/developer-guide/config-streams.html:
##########
@@ -70,11 +70,10 @@
<li><a class="reference internal" href="#optional-configuration-parameters" id="id6">Optional configuration parameters</a>
<ul>
<li><a class="reference internal" href="#acceptable-recovery-lag" id="id27">acceptable.recovery.lag</a></li>
- <li><a class="reference internal" href="#default-deserialization-exception-handler" id="id7">default.deserialization.exception.handler</a></li>
Review Comment:
I think this one was on the correct place alphabetically? Not sue why you moved it?
Cf my other comments below (not sure how to structure this best)
##########
docs/streams/developer-guide/config-streams.html:
##########
@@ -70,11 +70,10 @@
<li><a class="reference internal" href="#optional-configuration-parameters" id="id6">Optional configuration parameters</a>
<ul>
<li><a class="reference internal" href="#acceptable-recovery-lag" id="id27">acceptable.recovery.lag</a></li>
- <li><a class="reference internal" href="#default-deserialization-exception-handler" id="id7">default.deserialization.exception.handler</a></li>
<li><a class="reference internal" href="#default-key-serde" id="id8">default.key.serde</a></li>
- <li><a class="reference internal" href="#default-production-exception-handler" id="id24">default.production.exception.handler</a></li>
Review Comment:
as above
##########
docs/streams/developer-guide/config-streams.html:
##########
@@ -292,7 +294,7 @@ <h4><a class="toc-backref" href="#id23">num.standby.replicas</a><a class="header
set by the user or all serdes must be passed in explicitly (see also default.value.serde).</td>
<td><code class="docutils literal"><span class="pre">null</span></code></td>
</tr>
- <tr class="row-odd"><td>default.production.exception.handler</td>
+ <tr class="row-odd"><td>default.production.exception.handler (Deprecated. Use production.exception.handler instead.)</td>
Review Comment:
as above
##########
docs/streams/developer-guide/config-streams.html:
##########
@@ -281,7 +283,7 @@ <h4><a class="toc-backref" href="#id23">num.standby.replicas</a><a class="header
<td colspan="2">The frequency in milliseconds with which to save the position (offsets in source topics) of tasks.</td>
<td>30000 milliseconds (30 seconds)</td>
</tr>
- <tr class="row-odd"><td>default.deserialization.exception.handler</td>
+ <tr class="row-odd"><td>default.deserialization.exception.handler (Deprecated. Use deserialization.exception.handler instead.)</td>
Review Comment:
In this table, we should also insert a row for `deserialization.exception.handler`? Which can be basically a copy of this row?
Or do we think it would be redundant? For this case, we might want to rename this row to the new `deserialization.exception.handler`?
Not sure how to structure this nicely. Thoughts?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #17005:
URL: https://github.com/apache/kafka/pull/17005#issuecomment-2336528265
Merged to `trunk`. Congrats!
Thanks for the KIP and PR!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1741197901
##########
docs/streams/developer-guide/config-streams.html:
##########
@@ -70,9 +70,11 @@
<li><a class="reference internal" href="#optional-configuration-parameters" id="id6">Optional configuration parameters</a>
<ul>
<li><a class="reference internal" href="#acceptable-recovery-lag" id="id27">acceptable.recovery.lag</a></li>
- <li><a class="reference internal" href="#default-deserialization-exception-handler" id="id7">default.deserialization.exception.handler</a></li>
+ <li><a class="reference internal" href="#default-deserialization-exception-handler" id="id7">default.deserialization.exception.handler (deprecated)</a></li>
+ <li><a class="reference internal" href="#deserialization-exception-handler" id="id42">deserialization.exception.handler</a></li>
<li><a class="reference internal" href="#default-key-serde" id="id8">default.key.serde</a></li>
- <li><a class="reference internal" href="#default-production-exception-handler" id="id24">default.production.exception.handler</a></li>
+ <li><a class="reference internal" href="#default-production-exception-handler" id="id24">default.production.exception.handler (deprecated)</a></li>
+ <li><a class="reference internal" href="#production-exception-handler" id="id43">production.exception.handler</a></li>
Review Comment:
as above
##########
docs/streams/developer-guide/config-streams.html:
##########
@@ -70,9 +70,11 @@
<li><a class="reference internal" href="#optional-configuration-parameters" id="id6">Optional configuration parameters</a>
<ul>
<li><a class="reference internal" href="#acceptable-recovery-lag" id="id27">acceptable.recovery.lag</a></li>
- <li><a class="reference internal" href="#default-deserialization-exception-handler" id="id7">default.deserialization.exception.handler</a></li>
+ <li><a class="reference internal" href="#default-deserialization-exception-handler" id="id7">default.deserialization.exception.handler (deprecated)</a></li>
+ <li><a class="reference internal" href="#deserialization-exception-handler" id="id42">deserialization.exception.handler</a></li>
Review Comment:
We should also insert new rows with a detailed description (guess we can just c&p?) -- otherwise, we generate dangling links here.
##########
docs/streams/developer-guide/config-streams.html:
##########
@@ -70,9 +70,11 @@
<li><a class="reference internal" href="#optional-configuration-parameters" id="id6">Optional configuration parameters</a>
<ul>
<li><a class="reference internal" href="#acceptable-recovery-lag" id="id27">acceptable.recovery.lag</a></li>
- <li><a class="reference internal" href="#default-deserialization-exception-handler" id="id7">default.deserialization.exception.handler</a></li>
+ <li><a class="reference internal" href="#default-deserialization-exception-handler" id="id7">default.deserialization.exception.handler (deprecated)</a></li>
+ <li><a class="reference internal" href="#deserialization-exception-handler" id="id42">deserialization.exception.handler</a></li>
Review Comment:
We should insert in alphabetical order
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on PR #17005:
URL: https://github.com/apache/kafka/pull/17005#issuecomment-2310723601
@mjsax pls take a look
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1732464006
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImplTest.java:
##########
@@ -1158,6 +1159,75 @@ public void shouldSkipOnDeserializationErrorsWhenReprocessing() {
assertEquals(0, stateRestoreCallback.restored.size());
}
+ @SuppressWarnings("deprecation")
+ @Test
+ public void verifyExceptionHandlerAcceptNewConfigWhenBothArePresent() {
Review Comment:
Updated with warning when both are set.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1734149715
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1907,13 +1937,38 @@ public TimestampExtractor defaultTimestampExtractor() {
return getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
}
+ public DeserializationExceptionHandler deserializationExceptionHandler() {
+ if (getClass(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null &&
+ getClass(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null) {
+ log.warn("Both the deprecated and new config for deserialization exception handler are configured !!");
+ }
+ if (getClass(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null) {
+ return getConfiguredInstance(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
+ } else {
+ return defaultDeserializationExceptionHandler();
+ }
+ }
+
+ /**
+ * @deprecated as of kafka 4.0. Use deserializationExceptionHandler() instead
+ * @return DeserializationExceptionHandler
+ */
+ @Deprecated
@SuppressWarnings("WeakerAccess")
public DeserializationExceptionHandler defaultDeserializationExceptionHandler() {
return getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
}
+ public ProductionExceptionHandler productionExceptionHandler() {
+ if (getClass(PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG) != null) {
+ return getConfiguredInstance(PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG, ProductionExceptionHandler.class);
+ } else {
+ return defaultProductionExceptionHandler();
+ }
+ }
+
@SuppressWarnings("WeakerAccess")
- public ProductionExceptionHandler defaultProductionExceptionHandler() {
+ private ProductionExceptionHandler defaultProductionExceptionHandler() {
Review Comment:
Missed it from previous review. Updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1734161165
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -889,6 +901,11 @@ public class StreamsConfig extends AbstractConfig {
LogAndFailExceptionHandler.class.getName(),
Importance.MEDIUM,
DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC)
+ .define(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
+ Type.CLASS,
+ LogAndFailExceptionHandler.class.getName(),
Review Comment:
Pls ignore this comment. After using the originals().containsKey .. it is fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1735355654
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -918,12 +938,7 @@ public class StreamsConfig extends AbstractConfig {
Type.CLASS,
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)
+ PRODUCTION_EXCEPTION_HANDLER_CLASS_DOC)
Review Comment:
Both are the same so it does not really matter too much. Just feels itchy to use the new `_DOC` variable for the old config. -- And the fact that the new one is private should not matter?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1735364225
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1910,11 +1938,45 @@ public TimestampExtractor defaultTimestampExtractor() {
return getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
}
+ public DeserializationExceptionHandler deserializationExceptionHandler() {
+ if (originals().containsKey(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) &&
+ originals().containsKey(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG)) {
+ log.warn("Both the deprecated and new config for deserialization exception handler are configured. " +
+ "The deprecated one will be ignored.");
+ }
+ if (originals().containsKey(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG)) {
+ return getConfiguredInstance(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
+ } else {
+ return defaultDeserializationExceptionHandler();
+ }
+ }
+
+ /**
+ * @deprecated as of kafka 4.0; use {@link #deserializationExceptionHandler()} instead
+ */
+ @Deprecated
@SuppressWarnings("WeakerAccess")
public DeserializationExceptionHandler defaultDeserializationExceptionHandler() {
return getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
}
+ public ProductionExceptionHandler productionExceptionHandler() {
+ if (originals().containsKey(PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG) &&
+ originals().containsKey(DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG)) {
+ log.warn("Both the deprecated and new config for production exception handler are configured. " +
+ "The deprecated one will be ignored.");
+ }
+ if (originals().containsKey(PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG)) {
+ return getConfiguredInstance(PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG, ProductionExceptionHandler.class);
+ } else {
+ return defaultProductionExceptionHandler();
+ }
+ }
+
+ /**
+ * @deprecated as of kafka 4.0; use {@link #processingExceptionHandler()} instead
Review Comment:
```suggestion
* @deprecated since 4.0; use {@link #processingExceptionHandler()} instead
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1733654478
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -553,15 +553,35 @@ public class StreamsConfig extends AbstractConfig {
@SuppressWarnings("WeakerAccess")
public static final String CONNECTIONS_MAX_IDLE_MS_CONFIG = CommonClientConfigs.CONNECTIONS_MAX_IDLE_MS_CONFIG;
- /** {@code default.deserialization.exception.handler} */
+ /**
+ * {@code default.deserialization.exception.handler}
+ * @deprecated since 4.0.
+ * Use deserialization.exception.handler instead
+ */
@SuppressWarnings("WeakerAccess")
+ @Deprecated
public static final String DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG = "default.deserialization.exception.handler";
+ @Deprecated
public static final String DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC = "Exception handling class that implements the <code>org.apache.kafka.streams.errors.DeserializationExceptionHandler</code> interface.";
- /** {@code default.production.exception.handler} */
+ /** {@code deserialization.exception.handler} */
+ @SuppressWarnings("WeakerAccess")
+ public static final String DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG = "deserialization.exception.handler";
+ protected static final String DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC = "Exception handling class that implements the <code>org.apache.kafka.streams.errors.DeserializationExceptionHandler</code> interface.";
+
+ /**
+ * {@code default.production.exception.handler}
+ * @deprecated since 4.0.
+ * Use production.exception.handler instead
Review Comment:
as above
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1907,13 +1937,38 @@ public TimestampExtractor defaultTimestampExtractor() {
return getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
}
+ public DeserializationExceptionHandler deserializationExceptionHandler() {
+ if (getClass(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null &&
+ getClass(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null) {
+ log.warn("Both the deprecated and new config for deserialization exception handler are configured !!");
+ }
+ if (getClass(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null) {
+ return getConfiguredInstance(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
+ } else {
+ return defaultDeserializationExceptionHandler();
+ }
+ }
+
+ /**
+ * @deprecated as of kafka 4.0. Use deserializationExceptionHandler() instead
Review Comment:
```suggestion
* @deprecated as of kafka 4.0; use {@link #deserializationExceptionHandler()} instead
```
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1907,13 +1937,38 @@ public TimestampExtractor defaultTimestampExtractor() {
return getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
}
+ public DeserializationExceptionHandler deserializationExceptionHandler() {
+ if (getClass(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null &&
+ getClass(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null) {
+ log.warn("Both the deprecated and new config for deserialization exception handler are configured !!");
Review Comment:
```suggestion
log.warn("Both the deprecated and new config for deserialization exception handler are configured. The deprecated one will be ignored.");
```
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -918,12 +938,7 @@ public class StreamsConfig extends AbstractConfig {
Type.CLASS,
DefaultProductionExceptionHandler.class.getName(),
Importance.MEDIUM,
- DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_DOC)
- .define(PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG,
Review Comment:
Thanks for the side cleanup!
##########
streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java:
##########
@@ -1622,6 +1624,30 @@ public void testInvalidProcessingExceptionHandler() {
);
}
+ @Test
+ public void testDeserializationExceptionHandlerWhenOnlyNewConfigIsSet() {
+ props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndContinueExceptionHandler.class, streamsConfig.deserializationExceptionHandler().getClass());
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void testDeserializationExceptionHandlerWhenBothConfigsAreSet() {
+ props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+ props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndFailExceptionHandler.class);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndContinueExceptionHandler.class, streamsConfig.deserializationExceptionHandler().getClass());
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void testDeserializationExceptionHandlerWhenOnlyOldConfigIsSet() {
+ props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndFailExceptionHandler.class, streamsConfig.deserializationExceptionHandler().getClass());
Review Comment:
This one should expect `LogAndContinue` right? (I think that this test passing right now exposes the bug that you use `getClass()` right now, instead of `origins().containsKey()`
##########
streams/src/main/java/org/apache/kafka/streams/TopologyConfig.java:
##########
@@ -93,6 +95,11 @@ public class TopologyConfig extends AbstractConfig {
null,
Importance.MEDIUM,
DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC)
+ .define(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
Review Comment:
Seems this was not addressed yet.
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -553,15 +553,35 @@ public class StreamsConfig extends AbstractConfig {
@SuppressWarnings("WeakerAccess")
public static final String CONNECTIONS_MAX_IDLE_MS_CONFIG = CommonClientConfigs.CONNECTIONS_MAX_IDLE_MS_CONFIG;
- /** {@code default.deserialization.exception.handler} */
+ /**
+ * {@code default.deserialization.exception.handler}
+ * @deprecated since 4.0.
+ * Use deserialization.exception.handler instead
Review Comment:
Make one line:
```
@deprecated since 4.0; use {@link #DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG} instead
```
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -918,12 +938,7 @@ public class StreamsConfig extends AbstractConfig {
Type.CLASS,
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)
+ PRODUCTION_EXCEPTION_HANDLER_CLASS_DOC)
Review Comment:
Should this stay `DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_DOC`?
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1907,13 +1937,38 @@ public TimestampExtractor defaultTimestampExtractor() {
return getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
}
+ public DeserializationExceptionHandler deserializationExceptionHandler() {
+ if (getClass(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null &&
+ getClass(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null) {
+ log.warn("Both the deprecated and new config for deserialization exception handler are configured !!");
+ }
+ if (getClass(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null) {
+ return getConfiguredInstance(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
+ } else {
+ return defaultDeserializationExceptionHandler();
+ }
+ }
+
+ /**
+ * @deprecated as of kafka 4.0. Use deserializationExceptionHandler() instead
+ * @return DeserializationExceptionHandler
+ */
+ @Deprecated
@SuppressWarnings("WeakerAccess")
public DeserializationExceptionHandler defaultDeserializationExceptionHandler() {
return getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
}
+ public ProductionExceptionHandler productionExceptionHandler() {
+ if (getClass(PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG) != null) {
Review Comment:
This should work the same as for deserialization handler, including the check if both old and new configs are set?
##########
streams/src/main/java/org/apache/kafka/streams/TopologyConfig.java:
##########
@@ -223,11 +230,15 @@ public TopologyConfig(final String topologyName, final StreamsConfig globalAppCo
timestampExtractorSupplier = () -> globalAppConfigs.getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
}
- if (isTopologyOverride(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, topologyOverrides)) {
- deserializationExceptionHandlerSupplier = () -> getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
- log.info("Topology {} is overriding {} to {}", topologyName, DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, getClass(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG));
+ final String deserializationExceptionHandlerKey = getClass(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null ?
Review Comment:
as above: `getClass` -> `origins().containsKey()`
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1907,13 +1937,38 @@ public TimestampExtractor defaultTimestampExtractor() {
return getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
}
+ public DeserializationExceptionHandler deserializationExceptionHandler() {
+ if (getClass(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null &&
+ getClass(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null) {
+ log.warn("Both the deprecated and new config for deserialization exception handler are configured !!");
+ }
+ if (getClass(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null) {
Review Comment:
as above
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1907,13 +1937,38 @@ public TimestampExtractor defaultTimestampExtractor() {
return getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
}
+ public DeserializationExceptionHandler deserializationExceptionHandler() {
+ if (getClass(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null &&
+ getClass(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null) {
+ log.warn("Both the deprecated and new config for deserialization exception handler are configured !!");
+ }
+ if (getClass(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null) {
+ return getConfiguredInstance(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
+ } else {
+ return defaultDeserializationExceptionHandler();
+ }
+ }
+
+ /**
+ * @deprecated as of kafka 4.0. Use deserializationExceptionHandler() instead
+ * @return DeserializationExceptionHandler
+ */
+ @Deprecated
@SuppressWarnings("WeakerAccess")
public DeserializationExceptionHandler defaultDeserializationExceptionHandler() {
return getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
}
+ public ProductionExceptionHandler productionExceptionHandler() {
+ if (getClass(PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG) != null) {
+ return getConfiguredInstance(PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG, ProductionExceptionHandler.class);
+ } else {
+ return defaultProductionExceptionHandler();
+ }
+ }
+
@SuppressWarnings("WeakerAccess")
- public ProductionExceptionHandler defaultProductionExceptionHandler() {
+ private ProductionExceptionHandler defaultProductionExceptionHandler() {
Review Comment:
We cannot make public stuff private -- we need to first deprecate it
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1907,13 +1937,38 @@ public TimestampExtractor defaultTimestampExtractor() {
return getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
}
+ public DeserializationExceptionHandler deserializationExceptionHandler() {
+ if (getClass(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null &&
Review Comment:
`getClass` would return the default if not set. I think we would need to use
```
originals().containsKey(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG)
```
we should add a test if the log line is printed, too (cf helper `LogCaptureAppender.java`)
##########
streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java:
##########
@@ -1622,6 +1624,30 @@ public void testInvalidProcessingExceptionHandler() {
);
}
+ @Test
+ public void testDeserializationExceptionHandlerWhenOnlyNewConfigIsSet() {
+ props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndContinueExceptionHandler.class, streamsConfig.deserializationExceptionHandler().getClass());
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void testDeserializationExceptionHandlerWhenBothConfigsAreSet() {
+ props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+ props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndFailExceptionHandler.class);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndContinueExceptionHandler.class, streamsConfig.deserializationExceptionHandler().getClass());
Review Comment:
We can add an assertion for the WARN log line in this test.
##########
streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java:
##########
@@ -1622,6 +1624,30 @@ public void testInvalidProcessingExceptionHandler() {
);
}
+ @Test
+ public void testDeserializationExceptionHandlerWhenOnlyNewConfigIsSet() {
Review Comment:
```suggestion
public void shouldSetAndGetDeserializationExceptionHandlerWhenOnlyNewConfigIsSet() {
```
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1907,13 +1937,38 @@ public TimestampExtractor defaultTimestampExtractor() {
return getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
}
+ public DeserializationExceptionHandler deserializationExceptionHandler() {
+ if (getClass(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null &&
+ getClass(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null) {
+ log.warn("Both the deprecated and new config for deserialization exception handler are configured !!");
+ }
+ if (getClass(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) != null) {
+ return getConfiguredInstance(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
+ } else {
+ return defaultDeserializationExceptionHandler();
+ }
+ }
+
+ /**
+ * @deprecated as of kafka 4.0. Use deserializationExceptionHandler() instead
+ * @return DeserializationExceptionHandler
Review Comment:
I think we can omit this one
##########
streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java:
##########
@@ -1622,6 +1624,30 @@ public void testInvalidProcessingExceptionHandler() {
);
}
+ @Test
+ public void testDeserializationExceptionHandlerWhenOnlyNewConfigIsSet() {
+ props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndContinueExceptionHandler.class, streamsConfig.deserializationExceptionHandler().getClass());
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void testDeserializationExceptionHandlerWhenBothConfigsAreSet() {
+ props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+ props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndFailExceptionHandler.class);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndContinueExceptionHandler.class, streamsConfig.deserializationExceptionHandler().getClass());
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void testDeserializationExceptionHandlerWhenOnlyOldConfigIsSet() {
Review Comment:
```suggestion
public void shouldUseOldDeserializationExceptionHandlerWhenOnlyOldConfigIsSet() {
```
##########
streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java:
##########
@@ -1622,6 +1624,30 @@ public void testInvalidProcessingExceptionHandler() {
);
}
+ @Test
+ public void testDeserializationExceptionHandlerWhenOnlyNewConfigIsSet() {
+ props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndContinueExceptionHandler.class, streamsConfig.deserializationExceptionHandler().getClass());
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void testDeserializationExceptionHandlerWhenBothConfigsAreSet() {
Review Comment:
```suggestion
public void shouldUseNewDeserializationExceptionHandlerWhenBothConfigsAreSet() {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1734135862
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -918,12 +938,7 @@ public class StreamsConfig extends AbstractConfig {
Type.CLASS,
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)
+ PRODUCTION_EXCEPTION_HANDLER_CLASS_DOC)
Review Comment:
Because it is private already, aligned it with the config. I can change it back if you prefer to.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1733653722
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -889,6 +901,11 @@ public class StreamsConfig extends AbstractConfig {
LogAndFailExceptionHandler.class.getName(),
Importance.MEDIUM,
DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC)
+ .define(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
+ Type.CLASS,
+ LogAndFailExceptionHandler.class.getName(),
Review Comment:
Not sure if I understand the question. Can you elaborate?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1746023542
##########
docs/streams/developer-guide/config-streams.html:
##########
@@ -281,7 +283,7 @@ <h4><a class="toc-backref" href="#id23">num.standby.replicas</a><a class="header
<td colspan="2">The frequency in milliseconds with which to save the position (offsets in source topics) of tasks.</td>
<td>30000 milliseconds (30 seconds)</td>
</tr>
- <tr class="row-odd"><td>default.deserialization.exception.handler</td>
+ <tr class="row-odd"><td>default.deserialization.exception.handler (Deprecated. Use deserialization.exception.handler instead.)</td>
Review Comment:
yea it makes sense to add a new one, and in future we can delete this deprecated row totally.
Added new row and in alphabetical order
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1746023762
##########
docs/streams/developer-guide/config-streams.html:
##########
@@ -292,7 +294,7 @@ <h4><a class="toc-backref" href="#id23">num.standby.replicas</a><a class="header
set by the user or all serdes must be passed in explicitly (see also default.value.serde).</td>
<td><code class="docutils literal"><span class="pre">null</span></code></td>
</tr>
- <tr class="row-odd"><td>default.production.exception.handler</td>
+ <tr class="row-odd"><td>default.production.exception.handler (Deprecated. Use production.exception.handler instead.)</td>
Review Comment:
Added new row.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1748720586
##########
docs/streams/developer-guide/config-streams.html:
##########
@@ -292,7 +294,7 @@ <h4><a class="toc-backref" href="#id23">num.standby.replicas</a><a class="header
set by the user or all serdes must be passed in explicitly (see also default.value.serde).</td>
<td><code class="docutils literal"><span class="pre">null</span></code></td>
</tr>
- <tr class="row-odd"><td>default.production.exception.handler</td>
+ <tr class="row-odd"><td>default.production.exception.handler (Deprecated. Use production.exception.handler instead.)</td>
Review Comment:
Wen we insert a new row here, we need update each other row below with regard to `class="row-even"` vs ` class="row-odd"` so they alternate correctly (it's annoying, but otherwise the table wont' render as intended).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1735796925
##########
streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java:
##########
@@ -1622,6 +1626,72 @@ public void testInvalidProcessingExceptionHandler() {
);
}
+ @Test
+ public void shouldSetAndGetDeserializationExceptionHandlerWhenOnlyNewConfigIsSet() {
+ props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndContinueExceptionHandler.class, streamsConfig.deserializationExceptionHandler().getClass());
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void shouldUseNewDeserializationExceptionHandlerWhenBothConfigsAreSet() {
+ props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+ props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndFailExceptionHandler.class);
+
+ try (LogCaptureAppender streamsConfigLogs = LogCaptureAppender.createAndRegister(StreamsConfig.class)) {
+ streamsConfigLogs.setClassLogger(StreamsConfig.class, Level.WARN);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndContinueExceptionHandler.class, streamsConfig.deserializationExceptionHandler().getClass());
+
+ final long warningMessageWhenBothConfigsAreSet = streamsConfigLogs.getMessages().stream()
+ .filter(m -> m.contains("Both the deprecated and new config for deserialization exception handler are configured."))
+ .count();
+ assertEquals(1, warningMessageWhenBothConfigsAreSet);
+ }
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void shouldUseOldDeserializationExceptionHandlerWhenOnlyOldConfigIsSet() {
+ props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndContinueExceptionHandler.class, streamsConfig.deserializationExceptionHandler().getClass());
+ }
+
+ @Test
+ public void shouldSetAndGetProductionExceptionHandlerWhenOnlyNewConfigIsSet() {
+ props.put(StreamsConfig.PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG, DefaultProductionExceptionHandler.class);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(DefaultProductionExceptionHandler.class, streamsConfig.productionExceptionHandler().getClass());
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void shouldUseNewProductionExceptionHandlerWhenBothConfigsAreSet() {
+ props.put(StreamsConfig.PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG, DefaultProductionExceptionHandler.class);
+ props.put(StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG, ProductionExceptionHandler.class);
Review Comment:
Updated with ProductionExceptionHandlerMock, and also had to add a default constructor there now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on PR #17005:
URL: https://github.com/apache/kafka/pull/17005#issuecomment-2321453957
there are failing tests
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #17005:
URL: https://github.com/apache/kafka/pull/17005#issuecomment-2321839005
Yeah. Looks related. Can you take a look and push a fix?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on PR #17005:
URL: https://github.com/apache/kafka/pull/17005#issuecomment-2329029588
Pushed these changes. Was abit tricky.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #17005:
URL: https://github.com/apache/kafka/pull/17005
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1735848750
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -255,6 +255,7 @@ private static StreamsConfig createConfig(final String eosConfig, final String e
return createConfig(eosConfig, enforcedProcessingValue, deserializationExceptionHandler, LogAndFailProcessingExceptionHandler.class.getName());
}
+ @SuppressWarnings("deprecation")
Review Comment:
But after this change, I notice 3 other tests are failing. Strange, the config is not set properly, and originals are 0 in TopologyConfig. Not sure what am I missing. Can you pls take a look.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1732403377
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -555,12 +555,24 @@ public class StreamsConfig extends AbstractConfig {
/** {@code default.deserialization.exception.handler} */
@SuppressWarnings("WeakerAccess")
+ @Deprecated
public static final String DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG = "default.deserialization.exception.handler";
+ @Deprecated
public static final String DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC = "Exception handling class that implements the <code>org.apache.kafka.streams.errors.DeserializationExceptionHandler</code> interface.";
+ /** {@code deserialization.exception.handler} */
+ @SuppressWarnings("WeakerAccess")
+ public static final String DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG = "deserialization.exception.handler";
+
+ protected static final String DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC = "Exception handling class that implements the <code>org.apache.kafka.streams.errors.DeserializationExceptionHandler</code> interface.";
Review Comment:
As the config is referred in TopologyConfig too, thought `protected` is better
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1732453868
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -1907,16 +1929,42 @@ public TimestampExtractor defaultTimestampExtractor() {
return getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
}
+ public DeserializationExceptionHandler getDeserializationExceptionHandler() {
Review Comment:
Will update kip, and send an update.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1737574338
##########
streams/src/main/java/org/apache/kafka/streams/TopologyConfig.java:
##########
@@ -223,11 +230,15 @@ public TopologyConfig(final String topologyName, final StreamsConfig globalAppCo
timestampExtractorSupplier = () -> globalAppConfigs.getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
}
- if (isTopologyOverride(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, topologyOverrides)) {
- deserializationExceptionHandlerSupplier = () -> getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
- log.info("Topology {} is overriding {} to {}", topologyName, DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, getClass(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG));
+ final String deserializationExceptionHandlerKey = originals().containsKey(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) ?
Review Comment:
```suggestion
final String deserializationExceptionHandlerKey = globalAppConfigs.originals().containsKey(DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG) ?
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1735801452
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -255,6 +255,7 @@ private static StreamsConfig createConfig(final String eosConfig, final String e
return createConfig(eosConfig, enforcedProcessingValue, deserializationExceptionHandler, LogAndFailProcessingExceptionHandler.class.getName());
}
+ @SuppressWarnings("deprecation")
Review Comment:
Indeed, thanks. updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1746023542
##########
docs/streams/developer-guide/config-streams.html:
##########
@@ -281,7 +283,7 @@ <h4><a class="toc-backref" href="#id23">num.standby.replicas</a><a class="header
<td colspan="2">The frequency in milliseconds with which to save the position (offsets in source topics) of tasks.</td>
<td>30000 milliseconds (30 seconds)</td>
</tr>
- <tr class="row-odd"><td>default.deserialization.exception.handler</td>
+ <tr class="row-odd"><td>default.deserialization.exception.handler (Deprecated. Use deserialization.exception.handler instead.)</td>
Review Comment:
yea it makes sense to add a new one, and in future we can delete this deprecated row totally.
Added new row.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1746023762
##########
docs/streams/developer-guide/config-streams.html:
##########
@@ -292,7 +294,7 @@ <h4><a class="toc-backref" href="#id23">num.standby.replicas</a><a class="header
set by the user or all serdes must be passed in explicitly (see also default.value.serde).</td>
<td><code class="docutils literal"><span class="pre">null</span></code></td>
</tr>
- <tr class="row-odd"><td>default.production.exception.handler</td>
+ <tr class="row-odd"><td>default.production.exception.handler (Deprecated. Use production.exception.handler instead.)</td>
Review Comment:
Added new row in alphabetical order
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1742819290
##########
docs/streams/developer-guide/config-streams.html:
##########
@@ -83,6 +83,7 @@
<li><a class="reference internal" href="#probing-rebalance-interval-ms" id="id30">probing.rebalance.interval.ms</a></li>
<li><a class="reference internal" href="#processing-exception-handler" id="id41">processing.exception.handler</a></li>
<li><a class="reference internal" href="#processing-guarantee" id="id25">processing.guarantee</a></li>
+ <li><a class="reference internal" href="#production-exception-handler" id="id43">production.exception.handler</a></li>
Review Comment:
This is still adding a new link `<a ... href="#production-exception-handler" ...>` which is dangling...
I think we should update the HTML here to use `production-exception-handler` as new anchor point: https://kafka.apache.org/38/documentation/streams/developer-guide/config-streams.html#default-production-exception-handler
Ie, old and new should point to the same section which we update... (including the headline):
```
- <div class="section" id="default-production-exception-handler">
+ <div class="section" id="production-exception-handler">
```
And
```
- <span id="streams-developer-guide-peh"></span><h4><a class="toc-backref" href="#id24">default.production.exception.handler</a><a class="headerlink" href="#default-production-exception-handler" title="Permalink to this headline"></a></h4>
+ <span id="streams-developer-guide-peh"></span><h4><a class="toc-backref" href="#id24">production.exception.handler (deprecated: default.production.exception.handler)</a><a class="headerlink" href="#default-production-exception-handler" title="Permalink to this headline"></a></h4>
```
Same for the other config.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1737574959
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -255,6 +255,7 @@ private static StreamsConfig createConfig(final String eosConfig, final String e
return createConfig(eosConfig, enforcedProcessingValue, deserializationExceptionHandler, LogAndFailProcessingExceptionHandler.class.getName());
}
+ @SuppressWarnings("deprecation")
Review Comment:
Just needed a little bit of debugging... -> https://github.com/apache/kafka/pull/17005/files#r1737574338
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1733653179
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -555,12 +555,24 @@ public class StreamsConfig extends AbstractConfig {
/** {@code default.deserialization.exception.handler} */
@SuppressWarnings("WeakerAccess")
+ @Deprecated
public static final String DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG = "default.deserialization.exception.handler";
+ @Deprecated
public static final String DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC = "Exception handling class that implements the <code>org.apache.kafka.streams.errors.DeserializationExceptionHandler</code> interface.";
+ /** {@code deserialization.exception.handler} */
+ @SuppressWarnings("WeakerAccess")
+ public static final String DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG = "deserialization.exception.handler";
+
+ protected static final String DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC = "Exception handling class that implements the <code>org.apache.kafka.streams.errors.DeserializationExceptionHandler</code> interface.";
Review Comment:
For this case, let's use package private.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1735368052
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -255,6 +255,7 @@ private static StreamsConfig createConfig(final String eosConfig, final String e
return createConfig(eosConfig, enforcedProcessingValue, deserializationExceptionHandler, LogAndFailProcessingExceptionHandler.class.getName());
}
+ @SuppressWarnings("deprecation")
Review Comment:
We should updated the code to use `DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG` instead of `DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG` instead of adding a suppression
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1735368052
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java:
##########
@@ -255,6 +255,7 @@ private static StreamsConfig createConfig(final String eosConfig, final String e
return createConfig(eosConfig, enforcedProcessingValue, deserializationExceptionHandler, LogAndFailProcessingExceptionHandler.class.getName());
}
+ @SuppressWarnings("deprecation")
Review Comment:
Don't we need to update the code, too, to use `DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG` instead of `DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1735367470
##########
streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java:
##########
@@ -1622,6 +1626,72 @@ public void testInvalidProcessingExceptionHandler() {
);
}
+ @Test
+ public void shouldSetAndGetDeserializationExceptionHandlerWhenOnlyNewConfigIsSet() {
+ props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndContinueExceptionHandler.class, streamsConfig.deserializationExceptionHandler().getClass());
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void shouldUseNewDeserializationExceptionHandlerWhenBothConfigsAreSet() {
+ props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+ props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndFailExceptionHandler.class);
+
+ try (LogCaptureAppender streamsConfigLogs = LogCaptureAppender.createAndRegister(StreamsConfig.class)) {
+ streamsConfigLogs.setClassLogger(StreamsConfig.class, Level.WARN);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndContinueExceptionHandler.class, streamsConfig.deserializationExceptionHandler().getClass());
+
+ final long warningMessageWhenBothConfigsAreSet = streamsConfigLogs.getMessages().stream()
+ .filter(m -> m.contains("Both the deprecated and new config for deserialization exception handler are configured."))
+ .count();
+ assertEquals(1, warningMessageWhenBothConfigsAreSet);
+ }
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void shouldUseOldDeserializationExceptionHandlerWhenOnlyOldConfigIsSet() {
+ props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndContinueExceptionHandler.class, streamsConfig.deserializationExceptionHandler().getClass());
+ }
+
+ @Test
+ public void shouldSetAndGetProductionExceptionHandlerWhenOnlyNewConfigIsSet() {
+ props.put(StreamsConfig.PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG, DefaultProductionExceptionHandler.class);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(DefaultProductionExceptionHandler.class, streamsConfig.productionExceptionHandler().getClass());
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void shouldUseNewProductionExceptionHandlerWhenBothConfigsAreSet() {
+ props.put(StreamsConfig.PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG, DefaultProductionExceptionHandler.class);
+ props.put(StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG, ProductionExceptionHandler.class);
+
+ try (LogCaptureAppender streamsConfigLogs = LogCaptureAppender.createAndRegister(StreamsConfig.class)) {
+ streamsConfigLogs.setClassLogger(StreamsConfig.class, Level.WARN);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(DefaultProductionExceptionHandler.class, streamsConfig.productionExceptionHandler().getClass());
+
+ final long warningMessageWhenBothConfigsAreSet = streamsConfigLogs.getMessages().stream()
+ .filter(m -> m.contains("Both the deprecated and new config for production exception handler are configured."))
+ .count();
+ assertEquals(1, warningMessageWhenBothConfigsAreSet);
+ }
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void shouldUseOldProductionExceptionHandlerWhenOnlyOldConfigIsSet() {
+ props.put(StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG, DefaultProductionExceptionHandler.class);
Review Comment:
as above
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1735366368
##########
streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java:
##########
@@ -1622,6 +1626,72 @@ public void testInvalidProcessingExceptionHandler() {
);
}
+ @Test
+ public void shouldSetAndGetDeserializationExceptionHandlerWhenOnlyNewConfigIsSet() {
+ props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndContinueExceptionHandler.class, streamsConfig.deserializationExceptionHandler().getClass());
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void shouldUseNewDeserializationExceptionHandlerWhenBothConfigsAreSet() {
+ props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+ props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndFailExceptionHandler.class);
+
+ try (LogCaptureAppender streamsConfigLogs = LogCaptureAppender.createAndRegister(StreamsConfig.class)) {
+ streamsConfigLogs.setClassLogger(StreamsConfig.class, Level.WARN);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndContinueExceptionHandler.class, streamsConfig.deserializationExceptionHandler().getClass());
+
+ final long warningMessageWhenBothConfigsAreSet = streamsConfigLogs.getMessages().stream()
+ .filter(m -> m.contains("Both the deprecated and new config for deserialization exception handler are configured."))
+ .count();
+ assertEquals(1, warningMessageWhenBothConfigsAreSet);
+ }
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void shouldUseOldDeserializationExceptionHandlerWhenOnlyOldConfigIsSet() {
+ props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+ streamsConfig = new StreamsConfig(props);
+ assertEquals(LogAndContinueExceptionHandler.class, streamsConfig.deserializationExceptionHandler().getClass());
+ }
+
+ @Test
+ public void shouldSetAndGetProductionExceptionHandlerWhenOnlyNewConfigIsSet() {
+ props.put(StreamsConfig.PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG, DefaultProductionExceptionHandler.class);
Review Comment:
Well, given that the default is well, the default, setting the default does not give us any signal if it was picked up... Maybe use `ProductionExceptionHandlerMock` or implement a new dummy handler for this test itself.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on PR #17005:
URL: https://github.com/apache/kafka/pull/17005#issuecomment-2321425276
> Committed the fix.
>
> LGTM now, assuming build passes...
Great, thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on PR #17005:
URL: https://github.com/apache/kafka/pull/17005#issuecomment-2325915126
I have updated the upgrade section.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1742819658
##########
docs/streams/developer-guide/config-streams.html:
##########
@@ -70,11 +70,11 @@
<li><a class="reference internal" href="#optional-configuration-parameters" id="id6">Optional configuration parameters</a>
<ul>
<li><a class="reference internal" href="#acceptable-recovery-lag" id="id27">acceptable.recovery.lag</a></li>
- <li><a class="reference internal" href="#default-deserialization-exception-handler" id="id7">default.deserialization.exception.handler</a></li>
+ <li><a class="reference internal" href="#default-deserialization-exception-handler" id="id7">default.deserialization.exception.handler (deprecated since 4.0)</a></li>
<li><a class="reference internal" href="#default-key-serde" id="id8">default.key.serde</a></li>
- <li><a class="reference internal" href="#default-production-exception-handler" id="id24">default.production.exception.handler</a></li>
- <li><a class="reference internal" href="#timestamp-extractor" id="id15">default.timestamp.extractor</a></li>
+ <li><a class="reference internal" href="#default-production-exception-handler" id="id24">default.production.exception.handler (deprecated since 4.0)</a></li>
Review Comment:
```suggestion
<li><a class="reference internal" href="#production-exception-handler" id="id24">default.production.exception.handler (deprecated since 4.0)</a></li>
```
##########
docs/streams/developer-guide/config-streams.html:
##########
@@ -70,11 +70,11 @@
<li><a class="reference internal" href="#optional-configuration-parameters" id="id6">Optional configuration parameters</a>
<ul>
<li><a class="reference internal" href="#acceptable-recovery-lag" id="id27">acceptable.recovery.lag</a></li>
- <li><a class="reference internal" href="#default-deserialization-exception-handler" id="id7">default.deserialization.exception.handler</a></li>
+ <li><a class="reference internal" href="#default-deserialization-exception-handler" id="id7">default.deserialization.exception.handler (deprecated since 4.0)</a></li>
<li><a class="reference internal" href="#default-key-serde" id="id8">default.key.serde</a></li>
- <li><a class="reference internal" href="#default-production-exception-handler" id="id24">default.production.exception.handler</a></li>
- <li><a class="reference internal" href="#timestamp-extractor" id="id15">default.timestamp.extractor</a></li>
+ <li><a class="reference internal" href="#default-production-exception-handler" id="id24">default.production.exception.handler (deprecated since 4.0)</a></li>
Review Comment:
Cf my other comments
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1742819478
##########
docs/streams/developer-guide/config-streams.html:
##########
@@ -70,11 +70,11 @@
<li><a class="reference internal" href="#optional-configuration-parameters" id="id6">Optional configuration parameters</a>
<ul>
<li><a class="reference internal" href="#acceptable-recovery-lag" id="id27">acceptable.recovery.lag</a></li>
- <li><a class="reference internal" href="#default-deserialization-exception-handler" id="id7">default.deserialization.exception.handler</a></li>
+ <li><a class="reference internal" href="#default-deserialization-exception-handler" id="id7">default.deserialization.exception.handler (deprecated since 4.0)</a></li>
Review Comment:
```suggestion
<li><a class="reference internal" href="#deserialization-exception-handler" id="id7">default.deserialization.exception.handler (deprecated since 4.0)</a></li>
```
##########
docs/streams/developer-guide/config-streams.html:
##########
@@ -70,11 +70,11 @@
<li><a class="reference internal" href="#optional-configuration-parameters" id="id6">Optional configuration parameters</a>
<ul>
<li><a class="reference internal" href="#acceptable-recovery-lag" id="id27">acceptable.recovery.lag</a></li>
- <li><a class="reference internal" href="#default-deserialization-exception-handler" id="id7">default.deserialization.exception.handler</a></li>
+ <li><a class="reference internal" href="#default-deserialization-exception-handler" id="id7">default.deserialization.exception.handler (deprecated since 4.0)</a></li>
Review Comment:
Cf my other comments
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #17005:
URL: https://github.com/apache/kafka/pull/17005#discussion_r1742816586
##########
docs/streams/upgrade-guide.html:
##########
@@ -135,6 +135,12 @@ <h3 class="anchor-heading"><a id="streams_notable_changes" class="anchor-link"><
<h3><a id="streams_api_changes_400" href="#streams_api_changes_400">Streams API changes in 4.0.0</a></h3>
+ <p>
+ In this release two configs default.deserialization.exception.handler and default.production.exception.handler are deprecated, as they don't have any overwrites, which is described in
Review Comment:
```suggestion
In this release two configs <code>default.deserialization.exception.handler</code> and <code>default.production.exception.handler</code> are deprecated, as they don't have any overwrites, which is described in
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on PR #17005:
URL: https://github.com/apache/kafka/pull/17005#issuecomment-2324313809
Pushed a fix, also updated config-streams.html. ptal. thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on PR #17005:
URL: https://github.com/apache/kafka/pull/17005#issuecomment-2336808379
@mjsax Thank you for the great review and support.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16863 : Deprecate default exception handlers [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on PR #17005:
URL: https://github.com/apache/kafka/pull/17005#issuecomment-2336432784
Actually got the site-docs running locally now. Should have done this before. Nice, thanks. Learning a lot in this pr.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org