Retry-topics auto creation also creates the main topic

[jgslima]

In what version(s) of Spring for Apache Kafka are you seeing this issue?
2.9.1
Actually I already saw this behaviour since 2.8, not sure the oldest version that has this behaviour.

Describe the bug
When the autoCreateTopics feature of the Non-blocking retries main feature is activated, the framework also creates the main topic with the configurations (partitions and replicas) of the retry topics, even if the context has as NewTopic bean for the main topic with different configurations.

To Reproduce

Run the application below:

@SpringBootApplication
@EnableScheduling
@EnableKafkaRetryTopic
@Slf4j
public class SpringKafkaStudies {

	public static void main(String[] args) {
		SpringApplication.run(SpringKafkaStudies.class, args);
	}

	@Bean
	public NewTopic mainTopic() {
		return TopicBuilder
				.name("mytopic")
				.partitions(10)
				.replicas(2)
				.build();
	}

	@RetryableTopic(attempts = "3",
			backoff = @Backoff(delayExpression = "5000"),
			fixedDelayTopicStrategy = FixedDelayStrategy.SINGLE_TOPIC,
			autoCreateTopics = "true",
			numPartitions = "5",
			replicationFactor = "1")
	@KafkaListener(topics = "mytopic", concurrency = "3")
	public void nonBlockingRetriesListen(ConsumerRecord<String, String> record) {
		log.info("Received record: {}", record.value());
	}
	
}

Expected behavior
The main topic "mytopic" to be created with 10 partitions and 2 replicas, and the retry and DLT topic to be created with 5 partitions and 1 replicas.
But actually, the main topic "mytopic" is also created with 5 partitions and 1 replica.

Sample
See the code above.

[jgslima]

Indeed, as the default behaviour of spring-kafka, when sending the record to the retry or DLT topics, is to assume that the partition in those topics should be the same as the original topic, this means that the retry topics should have the same amount of partitions as the main one (although I think that the default behaviour of DeadLetterPublishingRecovererFactory.resolveTopicPartition() should be to not specify/resolve the partition, leaving this to the producer).

However the claim to have a distinct configuration of replication factor should be enough to expect that the framework might have at least some configuration to indicate that, the topics auto-creation feature of the retry mechanism should be done solely for the retry and DLT topics.

[garyrussell]

This is a bit of a hack, but it will serve as a work around until we can get this implemented:

@SpringBootApplication
public class Kgh2432Application {

	public static void main(String[] args) {
		SpringApplication.run(Kgh2432Application.class, args);
	}

	@KafkaListener(id = "kgh2432", topics = "kgh2432")
	@RetryableTopic
	void listen(String in) {
		System.out.println(in);
	}

	@Bean
	NewTopic topic() {
		return TopicBuilder.name("kgh2432").partitions(10).replicas(1).build();
	}

	@Bean
	ApplicationRunner runner(KafkaAdmin admin) {
		return args -> {
			Map<String, TopicDescription> topics = admin.describeTopics("kgh2432", "kgh2432-dlt");
			Assert.state(topics.get("kgh2432").partitions().size() == 10, "Wrong main partitions");
			Assert.state(topics.get("kgh2432-dlt").partitions().size() == 1, "Wrong DLT partitions");
		};
	}

}

@Component
class NewTopicWeeder implements SmartInitializingSingleton, Ordered {

	private final DefaultListableBeanFactory beanFactory;

	NewTopicWeeder (DefaultListableBeanFactory beanFactory) {
		this.beanFactory = beanFactory;
	}

	@Override
	public int getOrder() {
		return 0;
	}

	@Override
	public void afterSingletonsInstantiated() {
		this.beanFactory.destroySingleton("kgh2432-topicRegistrationBean");
	}

}