diff --git a/docs/src/main/asciidoc/kafka/kafka_overview.adoc b/docs/src/main/asciidoc/kafka/kafka_overview.adoc index a37e1eff0..008bdc098 100644 --- a/docs/src/main/asciidoc/kafka/kafka_overview.adoc +++ b/docs/src/main/asciidoc/kafka/kafka_overview.adoc @@ -367,7 +367,7 @@ The size of the batch is controlled by Kafka consumer properties `max.poll.recor Bear in mind that batch mode is not supported with `@StreamListener` - it only works with the newer functional programming model. IMPORTANT: Retry within the binder is not supported when using batch mode, so `maxAttempts` will be overridden to 1. -You can configure a `SeekToCurrentBatchErrorHandler` (using a `ListenerContainerCustomizer`) to achieve similar functionality to retry in the binder. +You can configure a `DefaultErrorHandler` (using a `ListenerContainerCustomizer`) to achieve similar functionality to retry in the binder. You can also use a manual `AckMode` and call `Ackowledgment.nack(index, sleep)` to commit the offsets for a partial batch and have the remaining records redelivered. Refer to the https://docs.spring.io/spring-kafka/docs/2.3.0.BUILD-SNAPSHOT/reference/html/#committing-offsets[Spring for Apache Kafka documentation] for more information about these techniques. diff --git a/docs/src/main/asciidoc/kafka/kafka_tips.adoc b/docs/src/main/asciidoc/kafka/kafka_tips.adoc index ee8f954e2..f7a71ce2a 100644 --- a/docs/src/main/asciidoc/kafka/kafka_tips.adoc +++ b/docs/src/main/asciidoc/kafka/kafka_tips.adoc @@ -157,14 +157,14 @@ In the case of `String` types, it does not make much sense, but for more complex Now when the consumer receives the data, let us assume that there is a bad record that causes a deserilziation errror, maybe someone passed an `Integer` instead of a `String` for example. In that case, if you don't do something in the application, the excption will be propagated through the chain and your application will exit eventually. -In order to handle this, you can add a `ListenerContainerCustomizer` `@Bean` that configures a `SeekToCurrentErrorHandler`. -This `SeekToCurrentErrorHandler` is configured with a `DeadLetterPublishingRecoverer`. +In order to handle this, you can add a `ListenerContainerCustomizer` `@Bean` that configures a `DefaultErrorHandler`. +This `DefaultErrorHandler` is configured with a `DeadLetterPublishingRecoverer`. We also need to configure an `ErrorHandlingDeserializer` for the consumer. That sounds like a lot of complex things, but in reality, it boils down to these 3 beans in this case. ``` @Bean - public ListenerContainerCustomizer> customizer(SeekToCurrentErrorHandler errorHandler) { + public ListenerContainerCustomizer> customizer(DefaultErrorHandler errorHandler) { return (container, dest, group) -> { container.setErrorHandler(errorHandler); }; @@ -173,8 +173,8 @@ That sounds like a lot of complex things, but in reality, it boils down to these ``` @Bean - public SeekToCurrentErrorHandler errorHandler(DeadLetterPublishingRecoverer deadLetterPublishingRecoverer) { - return new SeekToCurrentErrorHandler(deadLetterPublishingRecoverer); + public DefaultErrorHandler errorHandler(DeadLetterPublishingRecoverer deadLetterPublishingRecoverer) { + return new DefaultErrorHandler(deadLetterPublishingRecoverer); } ``` @@ -186,12 +186,12 @@ That sounds like a lot of complex things, but in reality, it boils down to these ``` Let us analyze each of them. -The first one is the `ListenerContainerCustomizer` bean that takes a `SeekToCurrentErrorHandler`. +The first one is the `ListenerContainerCustomizer` bean that takes a `DefaultErrorHandler`. The container is now customized with that particular error handler. You can learn more about container customization https://docs.spring.io/spring-cloud-stream/docs/current/reference/html/spring-cloud-stream.html#_advanced_consumer_configuration[here]. -The second bean is the `SeekToCurrentErrorHandler` that is configured with a publishing to a `DLT`. -See https://docs.spring.io/spring-kafka/docs/current/reference/html/#seek-to-current[here] for more details on `SeekToCurrentErrorHandler`. +The second bean is the `DefaultErrorHandler` that is configured with a publishing to a `DLT`. +See https://docs.spring.io/spring-kafka/docs/current/reference/html/#seek-to-current[here] for more details on `DefaultErrorHandler`. The third bean is the `DeadLetterPublishingRecoverer` that is ultimately responsible for sending to the `DLT`. By default, the `DLT` topic is named as the ORIGINAL_TOPIC_NAME.DLT.