Gracefully react to UNKNOWN_TOPIC_OR_PARTITION

[skliche]

Hi Spring Kafka team,

I was wondering if there is any way to react programmatically to the situation where a topic is not existing in kafka after creating a ConcurrentMessageListenerContainer. Looking through the code and documentation I did not find anything useful, so I hope you can help me out. Thanks in #advance.

The warning we see comes only from org.apache.kafka.clients.NetworkClient:

org.apache.kafka.clients.NetworkClient   : [Consumer clientId=domain-server-test-platform-data-events_90-1, groupId=domain-server_test-platform-data-events] Error while fetching metadata with correlation id {test-platform-data-events=UNKNOWN_TOPIC_OR_PARTITION} : {}

[garyrussell]

Unfortunately, the container starts poll() ing the consumer, without checking and we never get control back from the client code.

I think we should add a call to partitionsFor(topic) before starting to poll and fail to start the container if not present.

In the meantime, you could do your own check...

@Bean
public ApplicationRunner runner(ConsumerFactory<String, String> factory) {
	return args -> {
		Consumer<String, String> consumer = factory.createConsumer("foo", "bar");
		List<PartitionInfo> partitionsFor = consumer.partitionsFor("kgh696");
		System.out.println(partitionsFor);
		consumer.close();
	};
}

partitionsFor returns null if there are none.

[garyrussell]

Of course, you can also add a NewTopic bean to the context and Spring will create it for you during initialization:

@Bean
public NewTopic topic() {
	return new NewTopic("kgh696", 10, (short) 2);
}

[skliche]

Thank you very much for your quick answer. We will go for your approach, since another component is managing the topics explicitly.

[Wzy19930507]

@garyrussell hi, take a moment of your time, ask a piece of code

public class KafkaMessageListenerContainer {
		if (this.clientIdSuffix == null) {   // why need clientIdSuffix == null
			checkTopics();
		}
}

[garyrussell]

@Wzy19930507 Don't ask questions on issues - use Stack Overflow or the Discussions tab for Q&A.

The real comment explains it...

		if (this.clientIdSuffix == null) { // stand-alone container
			checkTopics();
		}

If this container is a child of a ConcurrentMessageListenerContainer there is no need to check the topics again because the parent already checked them.