GH-992: Fix SeekToCurrent zero retries

Fixes #992

The `SeekToCurrentErrorHandler` and `DefaultAfterRollbackProcessor`
always retried at least one time, even if `maxAttempts` was 1.

**cherry-pick to 2.2.x**

* Fix infinite retries

* Fix test class name and new Sonar issue.

Author: garyrussell

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

@@ -16,6 +16,7 @@
 
 package org.springframework.kafka.listener;
 
+import java.time.temporal.ValueRange;
 import java.util.function.BiConsumer;
 
 import org.apache.commons.logging.Log;
@@ -38,6 +39,8 @@ class FailedRecordTracker {
 
 	private final int maxFailures;
 
+	private final boolean noRetries;
+
 	FailedRecordTracker(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer, int maxFailures, Log logger) {
 		if (recoverer == null) {
 			this.recoverer = (r, t) -> logger.error("Max failures (" + maxFailures + ") reached for: " + r, t);
@@ -46,24 +49,34 @@ class FailedRecordTracker {
 			this.recoverer = recoverer;
 		}
 		this.maxFailures = maxFailures;
+		this.noRetries = ValueRange.of(0, 1).isValidIntValue(maxFailures);
 	}
 
 	boolean skip(ConsumerRecord<?, ?> record, Exception exception) {
+		if (this.noRetries) {
+			this.recoverer.accept(record, exception);
+			return true;
+		}
 		FailedRecord failedRecord = this.failures.get();
-		if (failedRecord == null || !failedRecord.getTopic().equals(record.topic())
-				|| failedRecord.getPartition() != record.partition() || failedRecord.getOffset() != record.offset()) {
+		if (this.maxFailures > 0 && (failedRecord == null || newFailure(record, failedRecord))) {
 			this.failures.set(new FailedRecord(record.topic(), record.partition(), record.offset()));
 			return false;
 		}
-		else {
-			if (this.maxFailures >= 0 && failedRecord.incrementAndGet() >= this.maxFailures) {
+		else if (this.maxFailures > 0 && failedRecord.incrementAndGet() >= this.maxFailures) {
 				this.recoverer.accept(record, exception);
 				return true;
-			}
+		}
+		else {
 			return false;
 		}
 	}
 
+	private boolean newFailure(ConsumerRecord<?, ?> record, FailedRecord failedRecord) {
+		return !failedRecord.getTopic().equals(record.topic())
+				|| failedRecord.getPartition() != record.partition()
+				|| failedRecord.getOffset() != record.offset();
+	}
+
 	void clearThreadState() {
 		this.failures.remove();
 	}

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

@@ -115,6 +115,8 @@
 public class KafkaMessageListenerContainer<K, V> // NOSONAR comment density
 		extends AbstractMessageListenerContainer<K, V> {
 
+	private static final String UNUSED = "unused";
+
 	private static final int DEFAULT_ACK_TIME = 5000;
 
 	private final AbstractMessageListenerContainer<K, V> container;
@@ -695,7 +697,7 @@ public void run() {
 				try {
 					pollAndInvoke();
 				}
-				catch (@SuppressWarnings("unused") WakeupException e) {
+				catch (@SuppressWarnings(UNUSED) WakeupException e) {
 					// Ignore, we're stopping
 				}
 				catch (NoOffsetForPartitionException nofpe) {
@@ -814,7 +816,7 @@ public void wrapUp() {
 					try {
 						this.consumer.unsubscribe();
 					}
-					catch (@SuppressWarnings("unused") WakeupException e) {
+					catch (@SuppressWarnings(UNUSED) WakeupException e) {
 						// No-op. Continue process
 					}
 				}
@@ -900,7 +902,7 @@ private void processAck(ConsumerRecord<K, V> record) {
 					try {
 						ackImmediate(record);
 					}
-					catch (@SuppressWarnings("unused") WakeupException e) {
+					catch (@SuppressWarnings(UNUSED) WakeupException e) {
 						// ignore - not polling
 					}
 				}
@@ -1040,7 +1042,7 @@ private RuntimeException doInvokeBatchListener(final ConsumerRecords<K, V> recor
 					throw er;
 				}
 			}
-			catch (@SuppressWarnings("unused") InterruptedException e) {
+			catch (@SuppressWarnings(UNUSED) InterruptedException e) {
 				Thread.currentThread().interrupt();
 			}
 			return null;
@@ -1539,7 +1541,7 @@ private void commitIfNecessary() {
 						this.consumer.commitAsync(commits, this.commitCallback);
 					}
 				}
-				catch (@SuppressWarnings("unused") WakeupException e) {
+				catch (@SuppressWarnings(UNUSED) WakeupException e) {
 					// ignore - not polling
 					this.logger.debug("Woken up during commit");
 				}

spring-kafka/src/test/java/org/springframework/kafka/listener/FailedRecordTrackerTests.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright 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.
+ * 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.listener;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.commons.logging.Log;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.junit.jupiter.api.Test;
+
+/**
+ * @author Gary Russell
+ * @since 2.2.5
+ *
+ */
+public class FailedRecordTrackerTests {
+
+	@Test
+	public void testNoRetries() {
+		AtomicBoolean recovered = new AtomicBoolean();
+		FailedRecordTracker tracker = new FailedRecordTracker((r, e) -> {
+			recovered.set(true);
+		}, 1, mock(Log.class));
+		ConsumerRecord<?, ?> record = new ConsumerRecord<>("foo", 0, 0L, "bar", "baz");
+		assertThat(tracker.skip(record, new RuntimeException())).isTrue();
+		assertThat(recovered.get()).isTrue();
+	}
+
+	@Test
+	public void testThreeRetries() {
+		AtomicBoolean recovered = new AtomicBoolean();
+		FailedRecordTracker tracker = new FailedRecordTracker((r, e) -> {
+			recovered.set(true);
+		}, 4, mock(Log.class));
+		ConsumerRecord<?, ?> record = new ConsumerRecord<>("foo", 0, 0L, "bar", "baz");
+		assertThat(tracker.skip(record, new RuntimeException())).isFalse();
+		assertThat(tracker.skip(record, new RuntimeException())).isFalse();
+		assertThat(tracker.skip(record, new RuntimeException())).isFalse();
+		assertThat(tracker.skip(record, new RuntimeException())).isTrue();
+		assertThat(recovered.get()).isTrue();
+	}
+
+}