Record doesn't get to DLT with FATAL exception and single-topic strategy

[tomazfernandes]

Currently if we throw a FATAL exception as per DLPR classification and we're using single topic strategy for fixed-delay non-blocking retries, the record doesn't get to the DLT.

To reproduce the issue, change the first listener in RetryTopicIntegrationTests to throw a ConversionException instead of the RuntimeException, the test will fail because the dlt countdown latch is never triggered.

This logic was introduced in DLPR to stop infinite loops for FATAL exceptions in DLT processing:

if (tp.topic().equals(record.topic()) && !getClassifier().classify(exception)) {
			this.logger.error("Recovery of " + ListenerUtils.recordToString(record, true)
					+ " skipped because not retryable exception " + exception.toString()
					+ " and the destination resolver routed back to the same topic");
			return;
		}

My suggestion is to have the FATAL classification in the DefaultDestinationTopicResolver, by having it extend the ExceptionClassifier class, and stop forwarding the record there, where we can know for sure we're in a DLT.

Also add a flag to DLPR to toggle the current behavior and set it to false in RetryableTopic's logic.

This solution should also address #2113, and open the way for us to solve #2116.

I'm already working on this and should have a PR open soon, if that's ok.

Let me know if you have any concerns or anything I might be missing.

Thanks

[garyrussell]

@tomazfernandes Thanks; I am off work until Tuesday, so we might not be able to get these changes into Tuesday's releases, unless it's a relatively small PR.