From 321b18f6c904eacea4ae573389bc53ca52fc42d9 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 21 Sep 2020 10:14:36 -0400 Subject: [PATCH] Add RecordMetadata to ProducerListener.onError When sending a record with no partition, and an error occurs, the `RecordMetadata` might contain the assigned partition if the error occurs after partition assignment; this may be useful information for the application. * Add @Nullable --- .../kafka/core/KafkaTemplate.java | 3 ++- .../support/CompositeProducerListener.java | 8 ++++++- .../support/LoggingProducerListener.java | 8 +++++-- .../kafka/support/ProducerListener.java | 24 ++++++++++++++++++- .../kafka/core/KafkaTemplateTests.java | 8 +++++-- src/reference/asciidoc/kafka.adoc | 3 ++- 6 files changed, 46 insertions(+), 8 deletions(-) diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java index 795b45cc..3739e0ac 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java @@ -385,6 +385,7 @@ public class KafkaTemplate implements KafkaOperations, ApplicationCo @Override public ListenableFuture> send(ProducerRecord record) { + Assert.notNull(record, "'record' cannot be null"); return doSend(record); } @@ -601,7 +602,7 @@ public class KafkaTemplate implements KafkaOperations, ApplicationCo } future.setException(new KafkaProducerException(producerRecord, "Failed to send", exception)); if (KafkaTemplate.this.producerListener != null) { - KafkaTemplate.this.producerListener.onError(producerRecord, exception); + KafkaTemplate.this.producerListener.onError(producerRecord, metadata, exception); } KafkaTemplate.this.logger.debug(exception, () -> "Failed to send: " + producerRecord); } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/CompositeProducerListener.java b/spring-kafka/src/main/java/org/springframework/kafka/support/CompositeProducerListener.java index d6183c45..dc91861f 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/CompositeProducerListener.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/CompositeProducerListener.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2019 the original author or authors. + * Copyright 2018-2020 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. @@ -74,8 +74,14 @@ public class CompositeProducerListener implements ProducerListener { } @Override + @Deprecated public void onError(ProducerRecord producerRecord, Exception exception) { this.delegates.forEach(d -> d.onError(producerRecord, exception)); } + @Override + public void onError(ProducerRecord producerRecord, RecordMetadata recordMetadata, Exception exception) { + this.delegates.forEach(d -> d.onError(producerRecord, recordMetadata, exception)); + } + } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/LoggingProducerListener.java b/spring-kafka/src/main/java/org/springframework/kafka/support/LoggingProducerListener.java index 65a9b065..a407c523 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/LoggingProducerListener.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/LoggingProducerListener.java @@ -18,8 +18,10 @@ package org.springframework.kafka.support; import org.apache.commons.logging.LogFactory; import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.clients.producer.RecordMetadata; import org.springframework.core.log.LogAccessor; +import org.springframework.lang.Nullable; import org.springframework.util.ObjectUtils; /** @@ -65,7 +67,7 @@ public class LoggingProducerListener implements ProducerListener { } @Override - public void onError(ProducerRecord record, Exception exception) { + public void onError(ProducerRecord record, @Nullable RecordMetadata recordMetadata, Exception exception) { LOGGER.error(exception, () -> { StringBuffer logOutput = new StringBuffer(); logOutput.append("Exception thrown when sending a message"); @@ -79,7 +81,9 @@ public class LoggingProducerListener implements ProducerListener { } logOutput.append(" to topic ").append(record.topic()); if (record.partition() != null) { - logOutput.append(" and partition ").append(record.partition()); + logOutput.append(" and partition ").append(recordMetadata != null + ? recordMetadata.partition() + : record.partition()); } logOutput.append(":"); return logOutput.toString(); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/ProducerListener.java b/spring-kafka/src/main/java/org/springframework/kafka/support/ProducerListener.java index 9507159b..ebd2447f 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/ProducerListener.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/ProducerListener.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2019 the original author or authors. + * Copyright 2015-2020 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. @@ -19,6 +19,8 @@ package org.springframework.kafka.support; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; +import org.springframework.lang.Nullable; + /** * Listener for handling outbound Kafka messages. Exactly one of its methods will be invoked, depending on whether * the write has been acknowledged or not. @@ -49,8 +51,28 @@ public interface ProducerListener { * Invoked after an attempt to send a message has failed. * @param producerRecord the failed record * @param exception the exception thrown + * @deprecated in favor of {@link #onError(ProducerRecord, RecordMetadata, Exception)}. */ + @Deprecated default void onError(ProducerRecord producerRecord, Exception exception) { } + /** + * Invoked after an attempt to send a message has failed. + * @param producerRecord the failed record + * @param recordMetadata The metadata for the record that was sent (i.e. the partition + * and offset). If an error occurred, metadata will contain only valid topic and maybe + * the partition. If the partition is not provided in the ProducerRecord and an error + * occurs before partition is assigned, then the partition will be set to + * RecordMetadata.UNKNOWN_PARTITION. + * @param exception the exception thrown + * @since 2.6.2 + */ + @SuppressWarnings("deprecation") + default void onError(ProducerRecord producerRecord, @Nullable RecordMetadata recordMetadata, + Exception exception) { + + onError(producerRecord, exception); + } + } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java b/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java index f531a641..aba4505a 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java @@ -53,6 +53,7 @@ import org.apache.kafka.clients.producer.RecordMetadata; import org.apache.kafka.common.Metric; import org.apache.kafka.common.MetricName; import org.apache.kafka.common.PartitionInfo; +import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.errors.TimeoutException; import org.apache.kafka.common.header.Header; import org.apache.kafka.common.serialization.Serializer; @@ -276,7 +277,9 @@ public class KafkaTemplateTests { } @Override - public void onError(ProducerRecord producerRecord, Exception exception) { + public void onError(ProducerRecord producerRecord, RecordMetadata metadata, + Exception exception) { + assertThat(producerRecord).isNotNull(); assertThat(exception).isNotNull(); onErrorDelegateCalls.incrementAndGet(); @@ -298,7 +301,8 @@ public class KafkaTemplateTests { //Drain the topic KafkaTestUtils.getSingleRecord(consumer, INT_KEY_TOPIC); pf.destroy(); - cpl.onError(records.get(0), new RuntimeException("x")); + cpl.onError(records.get(0), new RecordMetadata(new TopicPartition(INT_KEY_TOPIC, -1), 0L, 0L, 0L, 0L, 0, 0), + new RuntimeException("x")); assertThat(onErrorDelegateCalls.get()).isEqualTo(2); } diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index b1e16c22..1de1c43a 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -309,7 +309,8 @@ public interface ProducerListener { void onSuccess(ProducerRecord producerRecord, RecordMetadata recordMetadata); - void onError(ProducerRecord producerRecord, Exception exception); + void onError(ProducerRecord producerRecord, RecordMetadata recordMetadata, + Exception exception); } ----