GH-2826: Fix CommonDelegatingErrorHandler

Resolves #2826

The `addDelegate` method did not update the classifier, so it failed to
work when cause chain traversal is enabled.

**cherry-pick to 3.0.x, 2.9.x**

* Do not mutate the delegates field until after the validity check.
# Conflicts:
#	spring-kafka/src/main/java/org/springframework/kafka/listener/CommonDelegatingErrorHandler.java

Author: garyrussell

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021-2022 the original author or authors.
+ * Copyright 2021-2023 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.
@@ -70,15 +70,7 @@ public void setErrorHandlers(Map<Class<? extends Throwable>, CommonErrorHandler>
 		Assert.notNull(delegates, "'delegates' cannot be null");
 		this.delegates.clear();
 		this.delegates.putAll(delegates);
-		checkDelegates();
-		updateClassifier(delegates);
-	}
-
-	private void updateClassifier(Map<Class<? extends Throwable>, CommonErrorHandler> delegates) {
-		Map<Class<? extends Throwable>, Boolean> classifications = delegates.keySet().stream()
-			.map(commonErrorHandler -> Map.entry(commonErrorHandler, true))
-			.collect(Collectors.toMap(Entry::getKey, Entry::getValue));
-		this.classifier = new BinaryExceptionClassifier(classifications);
+		checkDelegatesAndUpdateClassifier(this.delegates);
 	}
 
 	/**
@@ -125,12 +117,17 @@ public void setAckAfterHandle(boolean ack) {
 	 * @param handler the handler.
 	 */
 	public void addDelegate(Class<? extends Throwable> throwable, CommonErrorHandler handler) {
-		this.delegates.put(throwable, handler);
-		checkDelegates();
+		Map<Class<? extends Throwable>, CommonErrorHandler> delegatesToCheck = new LinkedHashMap<>(this.delegates);
+		delegatesToCheck.put(throwable, handler);
+		checkDelegatesAndUpdateClassifier(delegatesToCheck);
+		this.delegates.clear();
+		this.delegates.putAll(delegatesToCheck);
 	}
 
 	@SuppressWarnings("deprecation")
-	private void checkDelegates() {
+	private void checkDelegatesAndUpdateClassifier(Map<Class<? extends Throwable>,
+			CommonErrorHandler> delegatesToCheck) {
+
 		boolean remainingRecords = this.defaultErrorHandler.remainingRecords();
 		boolean ackAfterHandle = this.defaultErrorHandler.isAckAfterHandle();
 		boolean seeksAfterHandling = this.defaultErrorHandler.seeksAfterHandling();
@@ -142,6 +139,14 @@ private void checkDelegates() {
 			Assert.isTrue(seeksAfterHandling == handler.seeksAfterHandling(),
 					"All delegates must return the same value when calling 'seeksAfterHandling()'");
 		});
+		updateClassifier(delegatesToCheck);
+	}
+
+	private void updateClassifier(Map<Class<? extends Throwable>, CommonErrorHandler> delegates) {
+		Map<Class<? extends Throwable>, Boolean> classifications = delegates.keySet().stream()
+			.map(commonErrorHandler -> Map.entry(commonErrorHandler, true))
+			.collect(Collectors.toMap(Entry::getKey, Entry::getValue));
+		this.classifier = new BinaryExceptionClassifier(classifications);
 	}
 
 	@Override

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

@@ -16,6 +16,7 @@
 
 package org.springframework.kafka.listener;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
@@ -31,6 +32,7 @@
 
 import org.springframework.kafka.KafkaException;
 import org.springframework.kafka.core.KafkaProducerException;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
 
 /**
  * Tests for {@link CommonDelegatingErrorHandler}.
@@ -134,7 +136,7 @@ void testDelegateForThrowableCauseIsAppliedWhenCauseTraversingIsEnabled() {
 	}
 
 	@Test
-	@SuppressWarnings("ConstantConditions")
+	@SuppressWarnings({ "ConstantConditions", "unchecked" })
 	void testDelegateForClassifiableThrowableCauseIsAppliedWhenCauseTraversingIsEnabled() {
 		var defaultHandler = mock(CommonErrorHandler.class);
 
@@ -147,6 +149,10 @@ void testDelegateForClassifiableThrowableCauseIsAppliedWhenCauseTraversingIsEnab
 		delegatingErrorHandler.setErrorHandlers(Map.of(
 			KafkaException.class, directCauseErrorHandler
 		));
+		delegatingErrorHandler.addDelegate(IllegalStateException.class, mock(CommonErrorHandler.class));
+		assertThat(KafkaTestUtils.getPropertyValue(delegatingErrorHandler, "classifier.classified", Map.class).keySet())
+				.contains(IllegalStateException.class);
+
 
 		delegatingErrorHandler.handleRemaining(exc, Collections.emptyList(), mock(Consumer.class),
 			mock(MessageListenerContainer.class));