future's get timeout is not working on KafkaTemplate

[ruxiao16]

Hi everyone,

I have a spring boot application set up with Spring Kafka. I did not explicitly configure the KafkaTemplate as most of the configuration is done via application property file. Recently I tried to add application logic to handle when Kafka cluster is not available so in my send function I did something like:

public void sendTdmSyncnToKafkaBlocking(final TdmSyncDto message) {
        try {
            tdmSyncTemplate.send(topicNameObj, message).get(5, TimeUnit.SECONDS);
        } catch (ExecutionException | InterruptedException | TimeoutException e) {
            handleFailure(message, e.getCause());
        }
    }

private void handleFailure(TdmSyncDto payload, Throwable e) {
        logger.info("The record : {} cannot be processed! caused by {}", payload, e);
    }

However, this is not working as expected because the timeoutException is only thrown after 60 seconds, which is the default time out value if i am mistaken.

[artembilan]

That's exactly how KafkaProducer works. It does not go to the async part until it fetches a metadata from the cluster. And this part is really done in the sync blocking way.
You may consider to change that option from one minute to something else:

    public static final String MAX_BLOCK_MS_CONFIG = "max.block.ms";
    private static final String MAX_BLOCK_MS_DOC = "The configuration controls how long the <code>KafkaProducer</code>'s <code>send()</code>, <code>partitionsFor()</code>, "
                                                    + "<code>initTransactions()</code>, <code>sendOffsetsToTransaction()</code>, <code>commitTransaction()</code> "
                                                    + "and <code>abortTransaction()</code> methods will block. "
                                                    + "For <code>send()</code> this timeout bounds the total time waiting for both metadata fetch and buffer allocation "
                                                    + "(blocking in the user-supplied serializers or partitioner is not counted against this timeout). "
                                                    + "For <code>partitionsFor()</code> this timeout bounds the time spent waiting for metadata if it is unavailable. "
                                                    + "The transaction-related methods always block, but may timeout if "
                                                    + "the transaction coordinator could not be discovered or did not respond within the timeout.";

[ruxiao16]

Thanks for the reply. So if I understand correctly, I will need to configure MAX_BLOCK_MS_CONFIG to be a smaller value than the 5 seconds I specified in KafkaTemplate.send("randomTopic", message).get(5, TimeUnit.SECONDS)? Could you please tell me which property is it mapped to in the profile.properties? Thank you.

[ruxiao16]

i have tried with spring.kafka.properties.max.block.ms= 10000 in the application property file based on this link but it complains about
2022-07-26 11:20:13.685 WARN 86556 --- [ main] o.a.k.clients.admin.AdminClientConfig : The configuration 'max.block.ms' was supplied but isn't a known config.

[artembilan]

The max.block.ms is a producer property. So, in Spring Boot it has to be configure like: spring.kafka.producer.properties[max.block.ms]=10000.
See more in docs: https://docs.spring.io/spring-boot/docs/current/reference/htmlsingle/#messaging.kafka.additional-properties