Infinite loop in Spring Kafka producer when sending message to an Invalid Topic

user_15318112 0
Viewed 1019

I'm working on a Spring boot application. The app is acts as a Kafka-producer in which producing/publishing messages is not very important. I'm using the spring-kafka dependency with the following configuration for Kafka

    @Bean
public ProducerFactory<String, Object> producerFactory() {
    Map<String, Object> configProps = new HashMap<>();

    configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConfig.getKafkaBrokers());
    configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
    configProps.put(ProducerConfig.RETRIES_CONFIG, 3);
    
    return new DefaultKafkaProducerFactory<>(configProps);
}

@Bean
public KafkaTemplate<String, Object> kafkaTemplate(ProducerFactory<String, Object> producerFactory) {
    return new KafkaTemplate<>(producerFactory);
}

When I send a message to a topic that doesnt exist. My application shows the following logs

17:18:45.187 [kafka-producer-network-thread | producer-1] WARN  org.apache.kafka.clients.NetworkClient - [] [Producer clientId=producer-1] Error while fetching metadata with correlation id 3 : {aksubd=UNKNOWN_TOPIC_OR_PARTITION}
17:18:45.347 [kafka-producer-network-thread | producer-1] WARN  org.apache.kafka.clients.NetworkClient - [] [Producer clientId=producer-1] Error while fetching metadata with correlation id 4 : {aksubd=UNKNOWN_TOPIC_OR_PARTITION}
17:18:45.522 [kafka-producer-network-thread | producer-1] WARN  org.apache.kafka.clients.NetworkClient - [] [Producer clientId=producer-1] Error while fetching metadata with correlation id 5 : {aksubd=UNKNOWN_TOPIC_OR_PARTITION}
17:18:45.687 [kafka-producer-network-thread | producer-1] WARN  org.apache.kafka.clients.NetworkClient - [] [Producer clientId=producer-1] Error while fetching metadata with correlation id 6 : {aksubd=UNKNOWN_TOPIC_OR_PARTITION}
17:18:45.864 [kafka-producer-network-thread | producer-1] WARN  org.apache.kafka.clients.NetworkClient - [] [Producer clientId=producer-1] Error while fetching metadata with correlation id 7 : {aksubd=UNKNOWN_TOPIC_OR_PARTITION}
17:18:46.035 [kafka-producer-network-thread | producer-1] WARN  org.apache.kafka.clients.NetworkClient - [] [Producer clientId=producer-1] Error while fetching metadata with correlation id 8 : {aksubd=UNKNOWN_TOPIC_OR_PARTITION}
17:18:46.203 [kafka-producer-network-thread | producer-1] WARN  org.apache.kafka.clients.NetworkClient - [] [Producer clientId=producer-1] Error while fetching metadata with correlation id 9 : {aksubd=UNKNOWN_TOPIC_OR_PARTITION}
17:18:46.378 [kafka-producer-network-thread | producer-1] WARN  org.apache.kafka.clients.NetworkClient - [] [Producer clientId=producer-1] Error while fetching metadata with correlation id 10 : {aksubd=UNKNOWN_TOPIC_OR_PARTITION}
17:18:46.546 [kafka-producer-network-thread | producer-1] WARN  org.apache.kafka.clients.NetworkClient - [] [Producer clientId=producer-1] Error while fetching metadata with correlation id 11 : {aksubd=UNKNOWN_TOPIC_OR_PARTITION}
17:18:46.715 [kafka-producer-network-thread | producer-1] WARN  org.apache.kafka.clients.NetworkClient - [] [Producer clientId=producer-1] Error while fetching metadata with correlation id 12 : {aksubd=UNKNOWN_TOPIC_OR_PARTITION}
17:18:46.888 [kafka-producer-network-thread | producer-1] WARN  org.apache.kafka.clients.NetworkClient - [] [Producer clientId=producer-1] Error while fetching metadata with correlation id 13 : {aksubd=UNKNOWN_TOPIC_OR_PARTITION}
17:18:47.055 [kafka-producer-network-thread | producer-1] WARN  org.apache.kafka.clients.NetworkClient - [] [Producer clientId=producer-1] Error while fetching metadata with correlation id 14 : {aksubd=UNKNOWN_TOPIC_OR_PARTITION}
17:18:47.233 [kafka-producer-network-thread | producer-1] WARN  org.apache.kafka.clients.NetworkClient - [] [Producer clientId=producer-1] Error while fetching metadata with correlation id 15 : {aksubd=UNKNOWN_TOPIC_OR_PARTITION}

The logs are in a infinite loop(I'm only showing the starting few logs). I deliberately gave an invalid topic "aksubd" that doesnt exist just to demonstrate the issue. I'm new to kafka and am wondering if this is normal behavior of Kafka as this may write unneccesary logs. I dont want the application to log infinitely. Is there a way to stop after a certain amount of retries that can be set in configuration and then throw an error onto the console? Note: I tried to set retries to 3 but it didnt work.

spring-kafka apache-kafka spring-boot spring java
1 Answers

Indeed, this is expected behavior.

If you have auto.create.topics.enable=true on the broker, then the topic would get created upon producer requests, but it is recommended to disable this, as clients could typo a topic name, then wonder where their data actually ends-up. Despite the name, retries doesn't apply for metadata requests, only producer batch requests.

If you want to check for a topic beforehand, use AdminClient to describe it, before producing to it.

user_23086830
Related