onError method not getting invoked on LoggingProducerListener exception

[shravani154]

I have a logic in onError method which needs to be invoked whenever there is a LoggingProducerListener exception. I are not able to catch this exception through try catch block also. I need to know why this exception occurs and how to handle/catch these exceptions. Below is the exception I am encountering.

2022-06-10 15:22:01.126 ERROR 1 --- [ad | producer-1] o.s.k.support.LoggingProducerListener : Exception thrown when sending a message with key='null' and payload='{123, 34, 116, 114, 97, 99, 101, 115, 34, 58, 32, 123, 34, 115, 101, 114, 118, 105, 99, 101, 78, 97,...' to topic

[artembilan]

What you are talking about is not a thrown exception, it is a log message under ERROR category.
See the logic of that LoggingProducerListener:

	public void onError(ProducerRecord<K, V> record, @Nullable RecordMetadata recordMetadata, Exception exception) {
		this.logger.error(exception, () -> {
			StringBuffer logOutput = new StringBuffer();
			logOutput.append("Exception thrown when sending a message");
			if (this.includeContents) {
				logOutput.append(" with key='")
					.append(keyOrValue(record.key()))
					.append("'")
					.append(" and payload='")
					.append(keyOrValue(record.value()))
					.append("'");
			}
			logOutput.append(" to topic ").append(record.topic());
			if (record.partition() != null) {
				logOutput.append(" and partition ").append(recordMetadata != null
						? recordMetadata.partition()
						: record.partition());
			}
			logOutput.append(":");
			return logOutput.toString();
		});
	}

So, probably better to inject your own ProducerListener with that respective onError() impl.
But again: that try..catch is done in the Kafka Client where they call a Callback provided, which in our cases just delegates to the provided ProducerListener.

In two words: the LoggingProducerListener does not throw exceptions to catch: it just logs them as ERRORs.

[artembilan]

Request parameters do not satisfy the configured policy.

Sounds like some Apache Kafka security config violation.

Consider to ask such a question in broader Kafka community.
This one is out of Spring for Apache Kafka scope.

[garyrussell]

Perhaps there are more clues in the server logs. If you can't figure it out from that, ask about this on Stack Overflow or on one the Kafka mailing lists: https://kafka.apache.org/contact

[shravani154]

Another question regarding LoggingProducerListener.
I am using cloud stream binders in my application and have multiple binders. I have my own Producer Listener in my application which logs on success and on error. In this scenario, I am not able to see onsuccess and on error messages if I use multiple binders. If I have a single binder the class is getting called, The issue is only when I have multiple binders. Is there any configuration I should set up?

Thanks In advance.

[garyrussell]

See the documentation - https://docs.spring.io/spring-cloud-stream/docs/current/reference/html/spring-cloud-stream.html#binder-customizer

You need to use a binder customizer instead when there are multiple binders.

[KhaiNguyenDuc]

Thanks, I have the same problem

[official-bet365]

The LoggingProducerListener operates asynchronously as part of Spring Kafka’s message production pipeline. When you send a message using a Kafka template, it typically doesn't block the main thread. Instead, the message is sent in the background, and any exceptions are handled by the producer listener configured in Spring Kafka.

The onError method in the listener is invoked when an exception occurs during the asynchronous send operation. Since this is outside the scope of your code's synchronous execution, try-catch blocks in your main logic won’t capture these exceptions.

[official-bet365]

To handle exceptions from the LoggingProducerListener, you can customize the producer listener or implement a new one. For example:

Custom Producer Listener: Implement your own ProducerListener to handle errors in the onError method:
public class CustomProducerListener<K, V> implements ProducerListener<K, V> {
@OverRide
public void onError(ProducerRecord<K, V> producerRecord, RecordMetadata recordMetadata, Exception exception) {
// Custom logic to handle the error
System.err.println("Error occurred: " + exception.getMessage());
// Add your specific error handling code here
}
}

[official-bet365]

Review Error Handling in Spring Kafka: Use Spring Kafka’s default exception handling mechanisms, such as retries or a dead-letter topic, to ensure robust processing.