RetryableTopic - provide a mechanism to suppress non-essential headers

[v-chernyshev]

Expected Behavior

I'd like to have a way to suppress the generation of the exception headers as they are quite verbose and drastically increase the size of the messages.

Current Behavior

Exception information headers are unconditionally added in this function:

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

Lines 363 to 364 in dc47e72

	private void addAndEnhanceHeaders(ConsumerRecord<?, ?> record, Exception exception,
	@Nullable DeserializationException vDeserEx, @Nullable DeserializationException kDeserEx, Headers headers) {

Context

In our application the initial attempt is quite likely to fail as we have to wait for data to be available in an external system that does not provide native notifications. The current retryable topics implementation adds a number of exception-related headers, including the full stacktrace information. When combined they increase the average message size from a couple hundred bytes to approximately 10 kilobytes. For us this overhead is hard to ignore.

Unfortunately, I do not see any way to hook into the header generation process. setHeadersFunction cannot be used as it is already customised by the recoverer factory:

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DeadLetterPublishingRecovererFactory.java

Line 139 in dc47e72

recoverer.setHeadersFunction((consumerRecord, e) -> addHeaders(consumerRecord, e, getAttempts(consumerRecord)));

The only workaround I can think of is to add a custom producer interceptor that strips the headers out of the records, but then the CPU resources are wasted on rendering the information that is almost immediately thrown away. It would be nice to avoid generating these headers in the first place.

[tomazfernandes]

Possibly related to this SO question.

But I think it might be worth it to have a dedicated option not to include the stacktrace in DeadLetterPublishingRecoverer.

[v-chernyshev]

Yeah, even if I get access to the original header function via reflection it won't help with suppressing the generation of the headers that are not strictly required. The reason for it is that this.headersFunction.apply is called after all the exception headers have already been set up.

[garyrussell]

I suggest we add something like

public interface HeaderCreator {

    void create(Headers kafkaHeaders, Exception exception, boolean isKey); 

}

(and a setter) with the default being this::addExceptionInfoHeaders

WDYT?

[v-chernyshev]

Something like setExceptionHeaderCreator would indeed allow me to provide a no-op implementation, which would solve this particular problem. Should this.headerNames.exceptionInfo be passed into the create method too? Given the comment here:

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

Lines 635 to 643 in 3c9e00f

	/**
	* Override this if you want different header names to be used
	* in the sent record.
	* @return the header names.
	* @since 2.7
	*/
	protected HeaderNames getHeaderNames() {
	return HeaderNames.Builder
	.original()

It is possible to replace the header names. Having programmatic access to this data could make the code a bit more generic. Although, admittedly, if the header names are replaced then you also know what they are.

[garyrussell]

Oh, because you might want to add just some of the headers (e.g. all except the full stack trace).

That makes sense.

Another option would be to add a BitSet to represent which exception headers to add, which would be simpler for most users, but still provide the hook to completely configure your own headers.

[tomazfernandes]

I think the solution seems great.

Just a thought, for the SO issue, considering we already have a setHeadersFunction for adding new headers in DLPR, we might have a similar method in DLPRF, with the function to be called in addHeaders.

[garyrussell]

Why would we need to add a setter to the factory, given that we already have a setter for a DLPR customizer? (Or maybe I misunderstand your comment).

[tomazfernandes]

DLPR already has a header configuration method - setHeadersFunction - but we are using it for the RT feature's header management. The way I suggested we would add additional complexity only in the feature's component, not having to add any additional complexity to DLPR to address what seems to be a RT's matter.

Also, I think the resulting API would be simpler, since the user would have an explicit method for handling this in the feature's component, instead of having to go through the customizer and knowing to use a different method than the setHeadersFunction there.

Of course, that's only a way of looking into this. Makes sense?

[garyrussell]

Good point; will add that too. PR coming soon...