diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ExceptionClassifier.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ExceptionClassifier.java
index 9846ad2695..756f6a0813 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ExceptionClassifier.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ExceptionClassifier.java
@@ -55,6 +55,16 @@ private static ExtendedBinaryExceptionClassifier configureDefaultClassifier() {
 		return new ExtendedBinaryExceptionClassifier(classified, true);
 	}
 
+	/**
+	 * By default, unmatched types classify as true. Call this method to make the default
+	 * false, and remove types explicitly classified as false. This should be called before
+	 * calling any of the classification modification methods.
+	 * @since 2.8.4
+	 */
+	public void defaultFalse() {
+		this.classifier = new ExtendedBinaryExceptionClassifier(new HashMap<>(), false);
+	}
+
 	/**
 	 * Return the exception classifier.
 	 * @return the classifier.
@@ -99,18 +109,37 @@ public void setClassifications(Map<Class<? extends Throwable>, Boolean> classifi
 	 * </ul>
 	 * All others will be retried.
 	 * @param exceptionTypes the exception types.
-	 * @see #removeNotRetryableException(Class)
+	 * @see #removeClassification(Class)
 	 * @see #setClassifications(Map, boolean)
 	 */
 	@SafeVarargs
 	@SuppressWarnings("varargs")
 	public final void addNotRetryableExceptions(Class<? extends Exception>... exceptionTypes) {
+		add(false, exceptionTypes);
+	}
+
+	/**
+	 * Add exception types that can be retried. Call this after {@link #defaultFalse()} to
+	 * specify those exception types that should be classified as true.
+	 * All others will not be retried.
+	 * @param exceptionTypes the exception types.
+	 * @see #removeClassification(Class)
+	 * @see #setClassifications(Map, boolean)
+	 * @since 2.8.4
+	 */
+	@SafeVarargs
+	@SuppressWarnings("varargs")
+	public final void addRetryableExceptions(Class<? extends Exception>... exceptionTypes) {
+		add(true, exceptionTypes);
+	}
+
+	private void add(boolean classified, Class<? extends Exception>... exceptionTypes) {
 		Assert.notNull(exceptionTypes, "'exceptionTypes' cannot be null");
 		Assert.noNullElements(exceptionTypes, "'exceptionTypes' cannot contain nulls");
 		for (Class<? extends Exception> exceptionType : exceptionTypes) {
 			Assert.isTrue(Exception.class.isAssignableFrom(exceptionType),
 					() -> "exceptionType " + exceptionType + " must be an Exception");
-			this.classifier.getClassified().put(exceptionType, false);
+			this.classifier.getClassified().put(exceptionType, classified);
 		}
 	}
 
@@ -125,13 +154,38 @@ public final void addNotRetryableExceptions(Class<? extends Exception>... except
 	 * <li>{@link NoSuchMethodException}</li>
 	 * <li>{@link ClassCastException}</li>
 	 * </ul>
-	 * All others will be retried.
+	 * All others will be retried, unless {@link #defaultFalse()} has been called.
 	 * @param exceptionType the exception type.
 	 * @return true if the removal was successful.
 	 * @see #addNotRetryableExceptions(Class...)
 	 * @see #setClassifications(Map, boolean)
+	 * @see #defaultFalse()
+	 * @deprecated in favor of {@link #removeClassification(Class)}
 	 */
+	@Deprecated
 	public boolean removeNotRetryableException(Class<? extends Exception> exceptionType) {
+		return this.removeClassification(exceptionType);
+	}
+
+	/**
+	 * Remove an exception type from the configured list. By default, the following
+	 * exceptions will not be retried:
+	 * <ul>
+	 * <li>{@link DeserializationException}</li>
+	 * <li>{@link MessageConversionException}</li>
+	 * <li>{@link ConversionException}</li>
+	 * <li>{@link MethodArgumentResolutionException}</li>
+	 * <li>{@link NoSuchMethodException}</li>
+	 * <li>{@link ClassCastException}</li>
+	 * </ul>
+	 * All others will be retried, unless {@link #defaultFalse()} has been called.
+	 * @param exceptionType the exception type.
+	 * @return true if the removal was successful.
+	 * @see #addNotRetryableExceptions(Class...)
+	 * @see #setClassifications(Map, boolean)
+	 * @since 2.8.4
+	 */
+	public boolean removeClassification(Class<? extends Exception> exceptionType) {
 		return this.classifier.getClassified().remove(exceptionType);
 	}
 
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DeadLetterPublishingRecovererFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DeadLetterPublishingRecovererFactory.java
index 33b04f1bcf..fa3645558b 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DeadLetterPublishingRecovererFactory.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DeadLetterPublishingRecovererFactory.java
@@ -143,7 +143,7 @@ protected DeadLetterPublishingRecoverer.HeaderNames getHeaderNames() {
 		recoverer.setSkipSameTopicFatalExceptions(false);
 		this.recovererCustomizer.accept(recoverer);
 		this.fatalExceptions.forEach(recoverer::addNotRetryableExceptions);
-		this.nonFatalExceptions.forEach(recoverer::removeNotRetryableException);
+		this.nonFatalExceptions.forEach(recoverer::removeClassification);
 		return recoverer;
 	}
 
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/DeadLetterPublishingRecovererTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/DeadLetterPublishingRecovererTests.java
index cfc4fda86f..056285ecd4 100644
--- a/spring-kafka/src/test/java/org/springframework/kafka/listener/DeadLetterPublishingRecovererTests.java
+++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/DeadLetterPublishingRecovererTests.java
@@ -559,7 +559,7 @@ void noCircularRoutingIfFatal() {
 		recoverer.addNotRetryableExceptions(IllegalStateException.class);
 		recoverer.accept(record, new IllegalStateException());
 		verify(template, never()).send(any(ProducerRecord.class));
-		recoverer.removeNotRetryableException(IllegalStateException.class);
+		recoverer.removeClassification(IllegalStateException.class);
 		recoverer.setFailIfSendResultIsError(false);
 		recoverer.accept(record, new IllegalStateException());
 		verify(template).send(any(ProducerRecord.class));
@@ -580,7 +580,7 @@ void doNotSkipCircularFatalIfSet() {
 		recoverer.addNotRetryableExceptions(IllegalStateException.class);
 		recoverer.accept(record, new IllegalStateException());
 		verify(template, times(2)).send(any(ProducerRecord.class));
-		recoverer.removeNotRetryableException(IllegalStateException.class);
+		recoverer.removeClassification(IllegalStateException.class);
 		recoverer.setFailIfSendResultIsError(false);
 		recoverer.accept(record, new IllegalStateException());
 		verify(template, times(3)).send(any(ProducerRecord.class));