GH-1828: Use consumer record's partition for @RetryableTopic (#1829)

* GH-1828: Add tests for RetryableTopic with autoCreate=false

Signed-off-by: Deepesh <deepeshvrm1@gmail.com>

* GH-1828: Update DeadLetterPublishingRecovererFactory#resolveDestination

- Extract out a protected method to `resolveTopicPartition` for destination DLT
- Set the partition to the same partition as the original record

* GH-1828: Remove un-necessary mockito stubbing from tests

* GH-1828: Add javadoc for DeadLetterPublishingRecovererFactory.resolveTopicPartition

Author: deepesh-verma

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DeadLetterPublishingRecovererFactory.java

@@ -115,8 +115,24 @@ private TopicPartition resolveDestination(ConsumerRecord<?, ?> cr, Exception e)
 
 		return nextDestination.isNoOpsTopic()
 					? null
-					: new TopicPartition(nextDestination.getDestinationName(),
-				cr.partition() % nextDestination.getDestinationPartitions());
+					: resolveTopicPartition(cr, nextDestination);
+	}
+
+	/**
+	 * Creates and returns the {@link TopicPartition}, where the original record should be forwarded.
+	 * By default, it will use the partition same as original record's partition, in the next destination topic.
+	 *
+	 * <p>{@link DeadLetterPublishingRecoverer#checkPartition} has logic to check whether that partition exists,
+	 * and if it doesn't it sets -1, to allow the Producer itself to assign a partition to the record.</p>
+	 *
+	 * <p>Subclasses can inherit from this method to override the implementation, if necessary.</p>
+	 *
+	 * @param cr The original {@link ConsumerRecord}, which is to be forwarded to DLT
+	 * @param nextDestination The next {@link DestinationTopic}, where the consumerRecord is to be forwarded
+	 * @return An instance of {@link TopicPartition}, specifying the topic and partition, where the cr is to be sent
+	 */
+	protected TopicPartition resolveTopicPartition(final ConsumerRecord<?, ?> cr, final DestinationTopic nextDestination) {
+		return new TopicPartition(nextDestination.getDestinationName(), cr.partition());
 	}
 
 	private int getAttempts(ConsumerRecord<?, ?> consumerRecord) {

spring-kafka/src/test/java/org/springframework/kafka/retrytopic/DeadLetterPublishingRecovererFactoryTests.java

@@ -102,7 +102,6 @@ void shouldSendMessage() {
 		given(destinationTopicResolver.resolveDestinationTopic(testTopic, 1, e, failureTimestamp)).willReturn(destinationTopic);
 		given(destinationTopic.isNoOpsTopic()).willReturn(false);
 		given(destinationTopic.getDestinationName()).willReturn(testRetryTopic);
-		given(destinationTopic.getDestinationPartitions()).willReturn(3);
 		given(destinationTopicResolver.getDestinationTopicByName(testRetryTopic)).willReturn(destinationTopic);
 		given(destinationTopic.getDestinationDelay()).willReturn(1000L);
 		willReturn(this.kafkaOperations).given(destinationTopic).getKafkaOperations();
@@ -145,7 +144,6 @@ void shouldIncreaseAttempts() {
 				.willReturn(destinationTopic);
 		given(destinationTopic.isNoOpsTopic()).willReturn(false);
 		given(destinationTopic.getDestinationName()).willReturn(testRetryTopic);
-		given(destinationTopic.getDestinationPartitions()).willReturn(1);
 		given(destinationTopicResolver.getDestinationTopicByName(testRetryTopic)).willReturn(destinationTopic);
 		willReturn(kafkaOperations).given(destinationTopic).getKafkaOperations();
 		given(kafkaOperations.send(any(ProducerRecord.class))).willReturn(listenableFuture);
@@ -175,7 +173,6 @@ void shouldAddOriginalTimestampHeader() {
 		given(destinationTopicResolver.resolveDestinationTopic(testTopic, 1, e, originalTimestamp)).willReturn(destinationTopic);
 		given(destinationTopic.isNoOpsTopic()).willReturn(false);
 		given(destinationTopic.getDestinationName()).willReturn(testRetryTopic);
-		given(destinationTopic.getDestinationPartitions()).willReturn(1);
 		given(destinationTopicResolver.getDestinationTopicByName(testRetryTopic)).willReturn(destinationTopic);
 		willReturn(this.kafkaOperations).given(destinationTopic).getKafkaOperations();
 		given(kafkaOperations.send(any(ProducerRecord.class))).willReturn(listenableFuture);
@@ -206,7 +203,6 @@ void shouldNotReplaceOriginalTimestampHeader() {
 		given(destinationTopicResolver.resolveDestinationTopic(testTopic, 1, e, timestamp)).willReturn(destinationTopic);
 		given(destinationTopic.isNoOpsTopic()).willReturn(false);
 		given(destinationTopic.getDestinationName()).willReturn(testRetryTopic);
-		given(destinationTopic.getDestinationPartitions()).willReturn(1);
 		given(destinationTopicResolver.getDestinationTopicByName(testRetryTopic)).willReturn(destinationTopic);
 		willReturn(this.kafkaOperations).given(destinationTopic).getKafkaOperations();
 		given(kafkaOperations.send(any(ProducerRecord.class))).willReturn(listenableFuture);