GH-2400: ReplyingKT Improve CorrelationId Logging

Resolves #2400

Use a hex string instead of a `BigInteger` in `toString`. Also, if the
bytes are the length of a UUID, use the standard representation thereof.

**cherry-pick to 2.9.x**

spring-kafka/src/main/java/org/springframework/kafka/requestreply/CorrelationKey.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,23 +16,25 @@
 
 package org.springframework.kafka.requestreply;
 
-import java.math.BigInteger;
 import java.util.Arrays;
 
 import org.springframework.util.Assert;
 
 /**
  * Wrapper for byte[] that can be used as a hash key. We could have used BigInteger
- * instead but this wrapper is much less expensive. We do use a BigInteger in
- * {@link #toString()} though.
+ * instead but this wrapper is much less expensive.
  *
  * @author Gary Russell
  * @since 2.1.3
  */
 public final class CorrelationKey {
 
+	private static final char[] HEX_ARRAY = "0123456789abcdef".toCharArray();
+
 	private final byte[] correlationId;
 
+	private String asString;
+
 	private volatile Integer hashCode;
 
 	public CorrelationKey(byte[] correlationId) { // NOSONAR array reference
@@ -74,9 +76,27 @@ public boolean equals(Object obj) {
 		return true;
 	}
 
+	private static String bytesToHex(byte[] bytes) {
+		boolean uuid = bytes.length == 16;
+		char[] hexChars = new char[bytes.length * 2 + (uuid ? 4 : 0)];
+		int i = 0;
+		for (int j = 0; j < bytes.length; j++) {
+			int v = bytes[j] & 0xFF;
+			hexChars[i++] = HEX_ARRAY[v >>> 4];
+			hexChars[i++] = HEX_ARRAY[v & 0x0F];
+			if (uuid && (j == 3 || j == 5 || j == 7 || j == 9)) {
+				hexChars[i++] = '-';
+			}
+		}
+		return new String(hexChars);
+	}
+
 	@Override
 	public String toString() {
-		return "[" + new BigInteger(this.correlationId) + "]";
+		if (this.asString == null) {
+			this.asString = bytesToHex(this.correlationId);
+		}
+		return this.asString;
 	}
 
 }

spring-kafka/src/test/java/org/springframework/kafka/requestreply/CorrelationKeyTests.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.requestreply;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.jupiter.api.Test;
+
+/**
+ * @author Gary Russell
+ * @since 2.8.10
+ *
+ */
+public class CorrelationKeyTests {
+
+	@Test
+	void asString() {
+		CorrelationKey key = new CorrelationKey(new byte[16]);
+		assertThat(key.toString()).isEqualTo("00000000-0000-0000-0000-000000000000");
+		key = new CorrelationKey(new byte[10]);
+		assertThat(key.toString()).isEqualTo("00000000000000000000");
+	}
+
+}