More tracing and debug logging

- Add trace log for successfull send
- Add debug log for failed send (debug because the user code should react to the future)
- Add trace logging for records received (topic-partition@offset) (container)
- Add debug logging for records received (topic-partition@offset) (test utils)

Author: garyrussell

spring-kafka-test/src/main/java/org/springframework/kafka/test/utils/KafkaTestUtils.java

@@ -20,6 +20,7 @@
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -163,7 +164,12 @@ public static <K, V> ConsumerRecords<K, V> getRecords(Consumer<K, V> consumer, l
 		logger.debug("Polling...");
 		ConsumerRecords<K, V> received = consumer.poll(timeout);
 		if (logger.isDebugEnabled()) {
-			logger.debug("Received: " + received.count());
+			logger.debug("Received: " + received.count() + ", "
+					+ received.partitions().stream()
+							.flatMap(p -> received.records(p).stream())
+							// map to same format as send metadata toString()
+							.map(r -> r.topic() + "-" + r.partition() + "@" + r.offset())
+							.collect(Collectors.toList()));
 		}
 		assertThat(received).as("null received from consumer.poll()").isNotNull();
 		return received;

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java

@@ -346,6 +346,9 @@ public void onCompletion(RecordMetadata metadata, Exception exception) {
 							KafkaTemplate.this.producerListener.onSuccess(producerRecord.topic(),
 									producerRecord.partition(), producerRecord.key(), producerRecord.value(), metadata);
 						}
+						if (KafkaTemplate.this.logger.isTraceEnabled()) {
+							KafkaTemplate.this.logger.trace("Sent ok: " + producerRecord + ", metadata: " + metadata);
+						}
 					}
 					else {
 						future.setException(new KafkaProducerException(producerRecord, "Failed to send", exception));
@@ -356,6 +359,9 @@ public void onCompletion(RecordMetadata metadata, Exception exception) {
 									producerRecord.value(),
 									exception);
 						}
+						if (KafkaTemplate.this.logger.isDebugEnabled()) {
+							KafkaTemplate.this.logger.debug("Failed to send: " + producerRecord, exception);
+						}
 					}
 				}
 				finally {

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

@@ -32,6 +32,7 @@
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ScheduledFuture;
+import java.util.stream.Collectors;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -657,6 +658,13 @@ public void run() {
 					ConsumerRecords<K, V> records = this.consumer.poll(this.containerProperties.getPollTimeout());
 					if (records != null && this.logger.isDebugEnabled()) {
 						this.logger.debug("Received: " + records.count() + " records");
+						if (records.count() > 0 && this.logger.isTraceEnabled()) {
+							this.logger.trace(records.partitions().stream()
+								.flatMap(p -> records.records(p).stream())
+								// map to same format as send metadata toString()
+								.map(r -> r.topic() + "-" + r.partition() + "@" + r.offset())
+								.collect(Collectors.toList()));
+						}
 					}
 					if (records != null && records.count() > 0) {
 						if (this.containerProperties.getIdleEventInterval() != null) {