GH-1189: Clear offsetsInThisBatch on nack()

Author: garyrussell

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

@@ -2900,12 +2900,17 @@ public void nack(long sleep) {
 						"nack() can only be called on the consumer thread");
 				Assert.isTrue(sleep >= 0, "sleep cannot be negative");
 				ListenerConsumer.this.nackSleep = sleep;
-				ListenerConsumer.this.deferredOffsets.forEach((part, recs) -> recs.clear());
+				synchronized (ListenerConsumer.this) {
+					if (ListenerConsumer.this.offsetsInThisBatch != null) {
+						ListenerConsumer.this.offsetsInThisBatch.forEach((part, recs) -> recs.clear());
+						ListenerConsumer.this.deferredOffsets.forEach((part, recs) -> recs.clear());
+					}
+				}
 			}
 
 			@Override
 			public String toString() {
-				return "Acknowledgment for " + this.record;
+				return "Acknowledgment for " + ListenerUtils.recordToString(this.record, true);
 			}
 
 		}
@@ -2944,6 +2949,12 @@ public void nack(int index, long sleep) {
 				Assert.isTrue(index >= 0 && index < this.records.count(), "index out of bounds");
 				ListenerConsumer.this.nackIndex = index;
 				ListenerConsumer.this.nackSleep = sleep;
+				synchronized (ListenerConsumer.this) {
+					if (ListenerConsumer.this.offsetsInThisBatch != null) {
+						ListenerConsumer.this.offsetsInThisBatch.forEach((part, recs) -> recs.clear());
+						ListenerConsumer.this.deferredOffsets.forEach((part, recs) -> recs.clear());
+					}
+				}
 			}
 
 			@Override