From 2907f9cc51fc86f21900e8d203a6a7d7ee06051e Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 29 Jan 2019 14:56:06 -0500 Subject: [PATCH] GH-948: Enhance LEFException with group.id Resolves https://github.com/spring-projects/spring-kafka/issues/948 For example, this can be used by the `DeadLetterPublishingRecoverer`'s destination resolver to choose a topic based on the group in addition to the information in the consumer record. * Polishing - @Nullable on getGroupId(). --- .../springframework/kafka/KafkaException.java | 5 ++- .../KafkaMessageListenerContainer.java | 16 +++++++- .../ListenerExecutionFailedException.java | 41 +++++++++++++++++-- .../listener/SeekToCurrentErrorHandler.java | 3 +- .../kafka/support/SeekUtils.java | 3 +- .../EnableKafkaIntegrationTests.java | 3 ++ .../listener/SeekToCurrentRecovererTests.java | 10 +++-- src/reference/asciidoc/kafka.adoc | 3 ++ src/reference/asciidoc/whats-new.adoc | 2 + 9 files changed, 74 insertions(+), 12 deletions(-) diff --git a/spring-kafka/src/main/java/org/springframework/kafka/KafkaException.java b/spring-kafka/src/main/java/org/springframework/kafka/KafkaException.java index be919be7..f61ee086 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/KafkaException.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/KafkaException.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2019 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -17,6 +17,7 @@ package org.springframework.kafka; import org.springframework.core.NestedRuntimeException; +import org.springframework.lang.Nullable; /** * The Spring Kafka specific {@link NestedRuntimeException} implementation. @@ -31,7 +32,7 @@ public class KafkaException extends NestedRuntimeException { super(message); } - public KafkaException(String message, Throwable cause) { + public KafkaException(String message, @Nullable Throwable cause) { super(message, cause); } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java index 8554579e..09a6b179 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java @@ -1243,6 +1243,18 @@ public class KafkaMessageListenerContainer // NOSONAR comment density @SuppressWarnings(RAWTYPES) @Nullable Producer producer, Iterator> iterator, RuntimeException e) { + Exception toHandle = e; + if (toHandle instanceof ListenerExecutionFailedException) { + toHandle = new ListenerExecutionFailedException(toHandle.getMessage(), this.consumerGroupId, + toHandle.getCause()); + } + else { + /* + * TODO: in 2.3, wrap all exceptions (e.g. thrown by user implementations + * of MessageListener) in LEFE with groupId. @KafkaListeners always throw + * LEFE. + */ + } if (this.errorHandler instanceof RemainingRecordsErrorHandler) { if (producer == null) { processCommits(); @@ -1252,11 +1264,11 @@ public class KafkaMessageListenerContainer // NOSONAR comment density while (iterator.hasNext()) { records.add(iterator.next()); } - ((RemainingRecordsErrorHandler) this.errorHandler).handle(e, records, this.consumer, + ((RemainingRecordsErrorHandler) this.errorHandler).handle(toHandle, records, this.consumer, KafkaMessageListenerContainer.this.container); } else { - this.errorHandler.handle(e, record, this.consumer); + this.errorHandler.handle(toHandle, record, this.consumer); } if (producer != null) { ackCurrent(record, producer); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerExecutionFailedException.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerExecutionFailedException.java index c5da0581..90fdf15e 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerExecutionFailedException.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerExecutionFailedException.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2019 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -17,6 +17,7 @@ package org.springframework.kafka.listener; import org.springframework.kafka.KafkaException; +import org.springframework.lang.Nullable; /** * The listener specific {@link KafkaException} extension. @@ -26,12 +27,46 @@ import org.springframework.kafka.KafkaException; @SuppressWarnings("serial") public class ListenerExecutionFailedException extends KafkaException { + private final String groupId; + + /** + * Construct an instance with the provided properties. + * @param message the exception message. + */ public ListenerExecutionFailedException(String message) { - super(message); + this(message, null, null); } - public ListenerExecutionFailedException(String message, Throwable cause) { + /** + * Construct an instance with the provided properties. + * @param message the exception message. + * @param cause the cause. + */ + public ListenerExecutionFailedException(String message, @Nullable Throwable cause) { + this(message, null, cause); + } + + /** + * Construct an instance with the provided properties. + * @param message the exception message. + * @param groupId the container's group.id property. + * @param cause the cause. + * @since 2.2.4 + */ + public ListenerExecutionFailedException(String message, @Nullable String groupId, @Nullable Throwable cause) { super(message, cause); + this.groupId = groupId; + } + + /** + * Return the consumer group.id property of the container that threw this exception. + * @return the group id; may be null, but not when the exception is passed to an error + * handler by a listener container. + * @since 2.2.4 + */ + @Nullable + public String getGroupId() { + return this.groupId; } } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentErrorHandler.java index ccea7583..75429614 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentErrorHandler.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentErrorHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-2019 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -88,6 +88,7 @@ public class SeekToCurrentErrorHandler implements ContainerAwareErrorHandler { @Override public void handle(Exception thrownException, List> records, Consumer consumer, MessageListenerContainer container) { + if (!SeekUtils.doSeeks(records, consumer, thrownException, true, this.failureTracker::skip, logger)) { throw new KafkaException("Seek to current after exception", thrownException); } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/SeekUtils.java b/spring-kafka/src/main/java/org/springframework/kafka/support/SeekUtils.java index de015a7b..c20af542 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/SeekUtils.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/SeekUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2018 the original author or authors. + * Copyright 2018-2019 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -57,6 +57,7 @@ public final class SeekUtils { */ public static boolean doSeeks(List> records, Consumer consumer, Exception exception, boolean recoverable, BiPredicate, Exception> skipper, Log logger) { + Map partitions = new LinkedHashMap<>(); AtomicBoolean first = new AtomicBoolean(true); AtomicBoolean skipped = new AtomicBoolean(); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java index d48cb020..40670f92 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java @@ -692,6 +692,9 @@ public class EnableKafkaIntegrationTests { template.send("annotated32", 0, 1, "foobar"); assertThat(this.config.listen16ErrorLatch.await(30, TimeUnit.SECONDS)).isTrue(); assertThat(this.config.listen16Exception).isNotNull(); + assertThat(this.config.listen16Exception).isInstanceOf(ListenerExecutionFailedException.class); + assertThat(((ListenerExecutionFailedException) this.config.listen16Exception).getGroupId()) + .isEqualTo("converter.explicitGroupId"); assertThat(this.config.listen16Message).isEqualTo("foobar"); } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentRecovererTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentRecovererTests.java index 653c6dc8..96557211 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentRecovererTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentRecovererTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2018 the original author or authors. + * Copyright 2018-2019 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -72,7 +72,6 @@ public class SeekToCurrentRecovererTests { public void testMaxFailures() throws Exception { Map props = KafkaTestUtils.consumerProps("seekTestMaxFailures", "false", embeddedKafka); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); - props.put(ConsumerConfig.GROUP_ID_CONFIG, "group"); props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties(topic1); @@ -87,7 +86,7 @@ public class SeekToCurrentRecovererTests { containerProps.setMessageListener((MessageListener) message -> { data.set(message.value()); if (message.offset() == 0) { - throw new RuntimeException("fail for max failures"); + throw new ListenerExecutionFailedException("fail for max failures"); } latch.countDown(); }); @@ -96,12 +95,16 @@ public class SeekToCurrentRecovererTests { new KafkaMessageListenerContainer<>(cf, containerProps); container.setBeanName("testSeekMaxFailures"); final CountDownLatch recoverLatch = new CountDownLatch(1); + final AtomicReference failedGroupId = new AtomicReference<>(); DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template, (r, e) -> new TopicPartition(topic1DLT, r.partition())) { @Override public void accept(ConsumerRecord record, Exception exception) { super.accept(record, exception); + if (exception instanceof ListenerExecutionFailedException) { + failedGroupId.set(((ListenerExecutionFailedException) exception).getGroupId()); + } recoverLatch.countDown(); } @@ -122,6 +125,7 @@ public class SeekToCurrentRecovererTests { assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue(); assertThat(data.get()).isEqualTo("bar"); assertThat(recoverLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(failedGroupId.get()).isEqualTo("seekTestMaxFailures"); container.stop(); Consumer consumer = cf.createConsumer(); embeddedKafka.consumeFromAnEmbeddedTopic(consumer, topic1DLT); diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index 134ee0c2..04f36f91 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -2527,6 +2527,9 @@ It also, optionally, can be configured with a `BiFunction, By default, the dead-letter record is sent to a topic named `.DLT` (the original topic name suffixed with `.DLT`) and to the same partition as the original record. Therefore, when using the default resolver, the dead-letter topic must have at least as many partitions as the original topic. If the returned `TopicPartition` has a negative partition, the partition is not set in the `ProducerRecord` and so the partition will be selected by Kafka. +Starting with version 2.2.4, any `ListenerExcutionFailedException` (e.g. thrown when an exception is detected in a `@KafkaListener` method) will be enhanced with the `groupId` property. +This will allow the destination resolver to use this in addition to the information in the `ConsumerRecord` to select the dead letter topic. + The following is an example of wiring a custom destination resolver. ==== diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 8bde57d9..79e981bb 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -38,6 +38,8 @@ See <> for more information. The `DefaultAfterRollbackProcessor` and `SeekToCurrentErrorHandler` can now recover (skip) records that keep failing, and will do so after 10 failures, by default. They can be configured to publish failed records to a dead-letter topic. +Starting with version 2.2.4, the consumer's group id can be used while selecting the dead letter topic name. + See <>, <> and <> for more information. The `ConsumerStoppingEvent` has been added.