KafkaProducerMessageHandler to detect issues/errors coming from kakfa-client

[yevsh]

Sometimes there are "error" network/kafka cluster errors - that can be only detected when we see in logs something like:

"INFO","logger":"org.apache.kafka.clients.NetworkClient","message":"[Producer clientId=producer-1] Node 2 disconnected.

The problem KafkaTemplate/KafkaProducerMessageHandler can't identify it as error, and data can be lost.

Need some reliable solution for this.

[artembilan]

Since that is just an INFO from the Kafka Client, then there is indeed nothing we can do on our side: https://stackoverflow.com/questions/73823931/node-disconnected-errors-in-kafka-streams-api.

It is probably better to ask such a question on Apache Kafka community.

[garyrussell]

To avoid data loss, you can set sync and the thread will block until the send is successful or throws an exception, or you can configure a sendFailureChannel to asynchronously receive an error message for the failure.

spring-integration/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java

Lines 306 to 315 in 8b004e9

	/**
	* Set the failure channel. After a send failure, an
	* {@link org.springframework.messaging.support.ErrorMessage} will be sent
	* to this channel with a payload of a {@link KafkaSendFailureException} with the
	* failed message and cause.
	* @param sendFailureChannel the failure channel.
	*/
	public void setSendFailureChannel(MessageChannel sendFailureChannel) {
	this.sendFailureChannel = sendFailureChannel;
	}

spring-integration/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java

Lines 268 to 276 in 8b004e9

	/**
	* A {@code boolean} indicating if the {@link KafkaProducerMessageHandler}
	* should wait for the send operation results or not. Defaults to {@code false}.
	* In {@code sync} mode a downstream send operation exception will be re-thrown.
	* @param sync the send mode; async by default.
	*/
	public void setSync(boolean sync) {
	this.sync = sync;
	}

[yevsh]

did as suggested :

 handler.setSync(true);
 handler.setSendFailureChannelName("failureChannel");

Then, KafkaTemplate's doSend() , the exception throw was triggered: (2.8.11 - if it's important)

 catch (ExecutionException e) {
	 throw new KafkaException("Send failed", e.getCause()); // NOSONAR, stack trace
 }

with error:
Caused by: org.springframework.kafka.KafkaException: Send failed; nested exception is org.apache.kafka.common.errors.TimeoutException: Topic xxxxxxx not present in metadata after 10000 ms.

but message didn't move to failureChannel :

@ServiceActivator(inputChannel = "failureChannel")
    public void handleFailure(Message<?> message) {
}

[artembilan]

The logic there is like this:

	if (failureChannel != null || metadataChannel != null) {
			future.whenComplete((sendResult, exception) -> {

Which is applied to the Future returned by the KafkaTemplate.send():

sendFuture = this.kafkaTemplate.send(producerRecord);
...
processSendResult(message, producerRecord, sendFuture, getSendSuccessChannel());

But turns out you gort a direct exception from that .kafkaTemplate.send() not a Future:

if (sendFuture.isDone()) {
			try {
				sendFuture.get();
			}
			catch (InterruptedException e) {
				Thread.currentThread().interrupt();
				throw new KafkaException("Interrupted", e);
			}
			catch (ExecutionException e) {
				throw new KafkaException("Send failed", e.getCause()); // NOSONAR, stack trace
			}
		}

Therefore that failureChannel is not called.
You need to look into a RequestHandlerRetryAdvice to catch that KafkaException and retry. If that is a logic you are looking for...

[garyrussell]

Actually, with sync, the timeout is thrown by this code.

spring-integration/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java

Lines 710 to 731 in c9023d1

	if (this.sync || this.isGateway) {
	Long sendTimeout = this.sendTimeoutExpression.getValue(this.evaluationContext, message, Long.class);
	if (sendTimeout != null && sendTimeout <= this.deliveryTimeoutMsProperty) {
	this.logger.debug(() -> "'sendTimeout' increased to "
	+ (this.deliveryTimeoutMsProperty + this.timeoutBuffer)
	+ "ms; it must be greater than the 'delivery.timeout.ms' Kafka producer "
	+ "property to avoid false failures");
	sendTimeout = this.deliveryTimeoutMsProperty + this.timeoutBuffer;
	}
	if (sendTimeout == null || sendTimeout < 0) {
	future.get();
	}
	else {
	try {
	future.get(sendTimeout, TimeUnit.MILLISECONDS);
	}
	catch (TimeoutException te) {
	throw new MessageTimeoutException(message, "Timeout waiting for response from KafkaProducer", te);
	}
	}
	}
	}

However, you don't need to do both; sync for exceptions to be thrown to the caller async for receiving the error asynchronously.

Note that even with async, some exceptions can still be thrown to the caller by the code that @artembilan referenced (e.g. message too large exceptions).

[garyrussell]

However, you should have received the error message in that case (see the code just above that).

Did you configure a consumer on the failure channel?

[garyrussell]

My apologies to @artembilan, in this case, the exception is thrown by the code he cited and will be thrown to the caller, regardless of the sync setting. There is no failure message sent for this exception.

@artembilan Do you think we should add a try/catch (and rethrow) block around this code:

spring-integration/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java

Lines 517 to 530 in c9023d1

	RequestReplyFuture<K, V, Object> gatewayFuture = null;
	if (this.isGateway && (!preBuilt || producerRecord.headers().lastHeader(KafkaHeaders.REPLY_TOPIC) == null)) {
	producerRecord.headers().add(new RecordHeader(KafkaHeaders.REPLY_TOPIC, getReplyTopic(message)));
	gatewayFuture = ((ReplyingKafkaTemplate<K, V, Object>) this.kafkaTemplate).sendAndReceive(producerRecord);
	sendFuture = gatewayFuture.getSendFuture();
	}
	else {
	if (this.transactional && !this.kafkaTemplate.inTransaction() && !this.allowNonTransactional) {
	sendFuture = this.kafkaTemplate.executeInTransaction(template -> template.send(producerRecord));
	}
	else {
	sendFuture = this.kafkaTemplate.send(producerRecord);
	}
	}

So we consistently send the error message?

On the other hand, if it is a direct channel, the caller will directly get the exception; if async, the exception will go to an error channel (if configured).

[artembilan]

I think, for consistency, we need to fix KafkaTemplate and don't throw an exception, but rather call CompletableFuture.completeExceptionally(ex).
This way the consumer of that code would always have only one way to react to the output of send() method call - hook into returned CompletableFuture.

[garyrussell]

We can't change that behavior in a patch release. That's why I suggested catching (and rethrowing) the exception in the message handler, but also send the failed message to the failure channel (if configured).

[artembilan]

OK. That sounds like a plan, Gary.
Would you mind proposing a Pull Request?
Thanks

[garyrussell]

#8638

[yevsh]

@artembilan
So to conclude this discussion, should i do what suggested above with RequestHandlerRetryAdvice?

I don't need to do explicit retry - just some other logic.

[artembilan]

Sure! You can use any other MethodInterceptor to catch those exceptions: https://docs.spring.io/spring-integration/docs/current/reference/html/messaging-endpoints.html#message-handler-advice-chain

[yevsh]

Still something not correct.
on handler:

       handler.setSync(true);
       handler.setAdviceChain(Collections.singletonList( myAdvice()));

advice:

 private ExpressionEvaluatingRequestHandlerAdvice myAdvice() {
        ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();

        advice.setFailureChannelName("failureChannel");
    
        advice.setTrapException(false); //true
        return advice;
    }

not triggered

[artembilan]

It is not clear if your handler is declared as a bean.
There is a logic like this:

	@Override
	protected final void onInit() {
		super.onInit();
		initAdvisedRequestHandlerIfAny();
		doInit();
	}

which is called from the afterPropertiesSet().

[yevsh]

handler is a bean, what is not triggered is the @ServiceActivator of failureChannel when Exception is thrown.

I see now,

	private Object evaluateFailureExpression(Message<?> message, Exception exception) {

	     DestinationResolver<MessageChannel> channelResolver = getChannelResolver();

failed on
java.lang.IllegalArgumentException: 'beanFactory' must not be null

[artembilan]

Oh! Right. The ExpressionEvaluatingRequestHandlerAdvice instance must be configured as a bean.

[yevsh]

works good now. thanks.

Still feels a bit un intuitive adding advise just for catching and handling such errors instead of using directly

handler.setSendFailureChannelName("failureChannel");

but it's ok.

[artembilan]

That handler.setSendFailureChannelName("failureChannel"); will work when we fix KafkaTemplate as I explained before with a CompletableFuture.completeExceptionally(ex) comment.
However it is indeed intuitive adding advise with Spring Integration concepts.

[yevsh]

Yes, I understand that. I also understand that fix will take time .So untill that will use the suggested approach.