GH-1653: Fix resetStateOnExceptionChange

Resolves #1653

Need to unwrap the cause of `ListenerExecutionFailedException` to determine
if the exception type changed.

* Add gradle plugin repo

Author: garyrussell

build.gradle

@@ -2,6 +2,7 @@ buildscript {
 	ext.kotlinVersion = '1.4.20'
 	repositories {
 		mavenCentral()
+		maven { url 'https://plugins.gradle.org/m2' }
 		maven { url 'https://repo.spring.io/plugins-release' }
 	}
 	dependencies {

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

@@ -149,18 +149,24 @@ boolean skip(ConsumerRecord<?, ?> record, Exception exception) {
 	private FailedRecord getFailedRecordInstance(ConsumerRecord<?, ?> record, Exception exception,
 			Map<TopicPartition, FailedRecord> map, TopicPartition topicPartition) {
 
+		Exception realException = exception;
+		if (realException  instanceof ListenerExecutionFailedException
+				&& realException.getCause() instanceof Exception) {
+
+			realException = (Exception) realException.getCause();
+		}
 		FailedRecord failedRecord = map.get(topicPartition);
 		if (failedRecord == null || failedRecord.getOffset() != record.offset()
 				|| (this.resetStateOnExceptionChange
-						&& !exception.getClass().isInstance(failedRecord.getLastException()))) {
+						&& !realException.getClass().isInstance(failedRecord.getLastException()))) {
 
-			failedRecord = new FailedRecord(record.offset(), determineBackOff(record, exception).start());
+			failedRecord = new FailedRecord(record.offset(), determineBackOff(record, realException).start());
 			map.put(topicPartition, failedRecord);
 		}
 		else {
 			failedRecord.getDeliveryAttempts().incrementAndGet();
 		}
-		failedRecord.setLastException(exception);
+		failedRecord.setLastException(realException);
 		return failedRecord;
 	}
 

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

@@ -171,9 +171,9 @@ void exceptionChanges(boolean reset) {
 			.isSameAs(be1);
 		TopicPartitionOffset tpo = new TopicPartitionOffset("foo", 0, 0L);
 		assertThat(tracker.deliveryAttempt(tpo)).isEqualTo(2);
-		tracker.skip(record1, new IllegalStateException());
+		tracker.skip(record1, new ListenerExecutionFailedException("test", new IllegalStateException()));
 		assertThat(tracker.deliveryAttempt(tpo)).isEqualTo(3);
-		tracker.skip(record1, new IllegalArgumentException());
+		tracker.skip(record1, new ListenerExecutionFailedException("test", new IllegalArgumentException()));
 		if (reset) {
 			assertThat(tracker.deliveryAttempt(tpo)).isEqualTo(2);
 			assertThat(KafkaTestUtils.getPropertyValue(failures.get()