GH-2003: Fix RetryableTopic Delivery Attempts Hdr

Resolves #2003

**cherry-pick to 2.7.x**

Author: garyrussell

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

@@ -17,6 +17,7 @@
 package org.springframework.kafka.retrytopic;
 
 import java.math.BigInteger;
+import java.nio.ByteBuffer;
 import java.time.Instant;
 import java.util.function.Consumer;
 
@@ -137,17 +138,28 @@ protected TopicPartition resolveTopicPartition(final ConsumerRecord<?, ?> cr, fi
 
 	private int getAttempts(ConsumerRecord<?, ?> consumerRecord) {
 		Header header = consumerRecord.headers().lastHeader(RetryTopicHeaders.DEFAULT_HEADER_ATTEMPTS);
-		return header != null
-				? header.value()[0]
-				: 1;
+		if (header != null) {
+			byte[] value = header.value();
+			if (value.length == 1) { // backwards compatibility
+				return value[0];
+			}
+			else if (value.length == 4) {
+				return ByteBuffer.wrap(value).getInt();
+			}
+			else {
+				LOGGER.debug(() -> "Unexected size for " + RetryTopicHeaders.DEFAULT_HEADER_ATTEMPTS + " header: "
+						+ value.length);
+			}
+		}
+		return 1;
 	}
 
 	private Headers addHeaders(ConsumerRecord<?, ?> consumerRecord, Exception e, int attempts) {
 		Headers headers = new RecordHeaders();
 		byte[] originalTimestampHeader = getOriginalTimestampHeaderBytes(consumerRecord);
 		headers.add(RetryTopicHeaders.DEFAULT_HEADER_ORIGINAL_TIMESTAMP, originalTimestampHeader);
 		headers.add(RetryTopicHeaders.DEFAULT_HEADER_ATTEMPTS,
-				BigInteger.valueOf(attempts + 1).toByteArray());
+				ByteBuffer.wrap(new byte[4]).putInt(attempts + 1).array());
 		headers.add(RetryTopicHeaders.DEFAULT_HEADER_BACKOFF_TIMESTAMP,
 				BigInteger.valueOf(getNextExecutionTimestamp(consumerRecord, e, originalTimestampHeader))
 						.toByteArray());

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

@@ -25,6 +25,7 @@
 import static org.mockito.Mockito.times;
 
 import java.math.BigInteger;
+import java.nio.ByteBuffer;
 import java.time.Clock;
 import java.time.Instant;
 import java.time.LocalDateTime;
@@ -125,22 +126,22 @@ void shouldSendMessage() {
 		// assert headers
 		Header attemptsHeader = producerRecord.headers().lastHeader(RetryTopicHeaders.DEFAULT_HEADER_ATTEMPTS);
 		assertThat(attemptsHeader).isNotNull();
-		assertThat(attemptsHeader.value()[0]).isEqualTo(Integer.valueOf(2).byteValue());
+		assertThat(ByteBuffer.wrap(attemptsHeader.value()).getInt()).isEqualTo(2);
 		Header timestampHeader = producerRecord.headers().lastHeader(RetryTopicHeaders.DEFAULT_HEADER_BACKOFF_TIMESTAMP);
 		assertThat(timestampHeader).isNotNull();
 		assertThat(new BigInteger(timestampHeader.value()).longValue()).isEqualTo(failureTimestamp + 1000L);
 	}
 
 	@Test
-	void shouldIncreaseAttempts() {
+	void shouldIncreaseAttemptsInLegacyHeader() {
 
 		// setup
 		RuntimeException e = new RuntimeException();
 		ConsumerRecord consumerRecord = new ConsumerRecord(testTopic, 0, 0, key, value);
-		consumerRecord.headers().add(RetryTopicHeaders.DEFAULT_HEADER_ATTEMPTS, BigInteger.valueOf(1).toByteArray());
+		consumerRecord.headers().add(RetryTopicHeaders.DEFAULT_HEADER_ATTEMPTS, BigInteger.valueOf(127).toByteArray());
 		consumerRecord.headers().add(RetryTopicHeaders.DEFAULT_HEADER_ORIGINAL_TIMESTAMP, this.originalTimestampBytes);
 
-		given(destinationTopicResolver.resolveDestinationTopic(testTopic, 1, e, originalTimestamp))
+		given(destinationTopicResolver.resolveDestinationTopic(testTopic, 127, e, originalTimestamp))
 				.willReturn(destinationTopic);
 		given(destinationTopic.isNoOpsTopic()).willReturn(false);
 		given(destinationTopic.getDestinationName()).willReturn(testRetryTopic);
@@ -159,7 +160,41 @@ void shouldIncreaseAttempts() {
 		ProducerRecord producerRecord = producerRecordCaptor.getValue();
 		Header attemptsHeader = producerRecord.headers().lastHeader(RetryTopicHeaders.DEFAULT_HEADER_ATTEMPTS);
 		assertThat(attemptsHeader).isNotNull();
-		assertThat(attemptsHeader.value()[0]).isEqualTo(Integer.valueOf(2).byteValue());
+		assertThat(attemptsHeader.value().length).isEqualTo(4); // handled a legacy one byte header ok
+		assertThat(ByteBuffer.wrap(attemptsHeader.value()).getInt()).isEqualTo(128);
+	}
+
+	@Test
+	void shouldIncreaseAttemptsInNewHeader() {
+
+		// setup
+		RuntimeException e = new RuntimeException();
+		ConsumerRecord consumerRecord = new ConsumerRecord(testTopic, 0, 0, key, value);
+		consumerRecord.headers().add(RetryTopicHeaders.DEFAULT_HEADER_ATTEMPTS,
+				ByteBuffer.wrap(new byte[4]).putInt(127).array());
+		consumerRecord.headers().add(RetryTopicHeaders.DEFAULT_HEADER_ORIGINAL_TIMESTAMP, this.originalTimestampBytes);
+
+		given(destinationTopicResolver.resolveDestinationTopic(testTopic, 127, e, originalTimestamp))
+				.willReturn(destinationTopic);
+		given(destinationTopic.isNoOpsTopic()).willReturn(false);
+		given(destinationTopic.getDestinationName()).willReturn(testRetryTopic);
+		given(destinationTopicResolver.getDestinationTopicByName(testRetryTopic)).willReturn(destinationTopic);
+		willReturn(kafkaOperations).given(destinationTopic).getKafkaOperations();
+		given(kafkaOperations.send(any(ProducerRecord.class))).willReturn(listenableFuture);
+
+		DeadLetterPublishingRecovererFactory factory = new DeadLetterPublishingRecovererFactory(this.destinationTopicResolver);
+
+		// when
+		DeadLetterPublishingRecoverer deadLetterPublishingRecoverer = factory.create();
+		deadLetterPublishingRecoverer.accept(consumerRecord, e);
+
+		// then
+		then(kafkaOperations).should(times(1)).send(producerRecordCaptor.capture());
+		ProducerRecord producerRecord = producerRecordCaptor.getValue();
+		Header attemptsHeader = producerRecord.headers().lastHeader(RetryTopicHeaders.DEFAULT_HEADER_ATTEMPTS);
+		assertThat(attemptsHeader).isNotNull();
+		assertThat(attemptsHeader.value().length).isEqualTo(4);
+		assertThat(ByteBuffer.wrap(attemptsHeader.value()).getInt()).isEqualTo(128);
 	}
 
 	@Test