GH-691: Add CompositeProducerListener

Fixes #691

**cherry-pick to 2.1.x**

* Polishing - PR Comments

Improve test.

(cherry picked from commit 2355bda)

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/support/CompositeProducerListener.java

@@ -0,0 +1,76 @@
+/*
+ * Copyright 2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.support;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+
+import org.springframework.util.Assert;
+
+/**
+ * A {@link ProducerListener} that delegates to a collection of listeners.
+ *
+ * @param <K> the key type.
+ * @param <V> the value type.
+ * @author Gary Russell
+ *
+ * @since 2.1.6
+ *
+ */
+public class CompositeProducerListener<K, V> implements ProducerListener<K, V> {
+
+	private final List<ProducerListener<K, V>> delegates = new CopyOnWriteArrayList<>();
+
+	@SafeVarargs
+	public CompositeProducerListener(ProducerListener<K, V>... delegates) {
+		setDelegates(delegates);
+	}
+
+	@SafeVarargs
+	public final void setDelegates(ProducerListener<K, V>... delegates) {
+		Assert.notNull(delegates, "'delegates' cannot be null");
+		this.delegates.clear();
+		this.delegates.addAll(Arrays.asList(delegates));
+	}
+
+	protected List<ProducerListener<K, V>> getDelegates() {
+		return this.delegates;
+	}
+
+	public void addDelegate(ProducerListener<K, V> delegate) {
+		this.delegates.add(delegate);
+	}
+
+	public boolean removeDelegate(ProducerListener<K, V> delegate) {
+		return this.delegates.remove(delegate);
+	}
+
+	@Override
+	public void onSuccess(ProducerRecord<K, V> producerRecord, RecordMetadata recordMetadata) {
+		this.delegates.forEach(d -> d.onSuccess(producerRecord, recordMetadata));
+	}
+
+	@Override
+	public void onError(ProducerRecord<K, V> producerRecord, Exception exception) {
+		this.delegates.forEach(d -> d.onError(producerRecord, exception));
+	}
+
+}

spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java

@@ -23,11 +23,13 @@
 import static org.springframework.kafka.test.assertj.KafkaConditions.timestamp;
 import static org.springframework.kafka.test.assertj.KafkaConditions.value;
 
+import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.kafka.clients.consumer.Consumer;
@@ -49,6 +51,7 @@
 import org.junit.Test;
 
 import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.kafka.support.CompositeProducerListener;
 import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.kafka.support.ProducerListener;
@@ -242,23 +245,44 @@ public void withListener() throws Exception {
 		DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
 		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
 		template.setDefaultTopic(INT_KEY_TOPIC);
-		final CountDownLatch latch = new CountDownLatch(1);
-		template.setProducerListener(new ProducerListener<Integer, String>() {
+		final CountDownLatch latch = new CountDownLatch(2);
+		final List<ProducerRecord<Integer, String>> records = new ArrayList<>();
+		final List<RecordMetadata> meta = new ArrayList<>();
+		final AtomicInteger onErrorDelegateCalls = new AtomicInteger();
+		class PL implements ProducerListener<Integer, String> {
 
 			@Override
-			public void onSuccess(String topic, Integer partition, Integer key, String value,
-					RecordMetadata recordMetadata) {
+			public void onSuccess(ProducerRecord<Integer, String> record, RecordMetadata recordMetadata) {
+				records.add(record);
+				meta.add(recordMetadata);
 				latch.countDown();
 			}
 
-		});
+			@Override
+			public void onError(ProducerRecord<Integer, String> producerRecord, Exception exception) {
+				assertThat(producerRecord).isNotNull();
+				assertThat(exception).isNotNull();
+				onErrorDelegateCalls.incrementAndGet();
+			}
+
+		}
+		PL pl1 = new PL();
+		PL pl2 = new PL();
+		CompositeProducerListener<Integer, String> cpl = new CompositeProducerListener<>(pl1, pl2);
+		template.setProducerListener(cpl);
 		template.sendDefault("foo");
 		template.flush();
 		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(records.get(0).value()).isEqualTo("foo");
+		assertThat(records.get(1).value()).isEqualTo("foo");
+		assertThat(meta.get(0).topic()).isEqualTo(INT_KEY_TOPIC);
+		assertThat(meta.get(1).topic()).isEqualTo(INT_KEY_TOPIC);
 
 		//Drain the topic
 		KafkaTestUtils.getSingleRecord(consumer, INT_KEY_TOPIC);
 		pf.destroy();
+		cpl.onError(records.get(0), new RuntimeException("x"));
+		assertThat(onErrorDelegateCalls.get()).isEqualTo(2);
 	}
 
 	@Test