spring-projectsGH-1898: Fix OOM in LoggingProducerListener

Resolves spring-projects#1898 (comment)

Large `byte[]` payloads could cause an OOM.

**cherry-pick to 2.7.x, 2.6.x, 2.5.x**

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/support/LoggingProducerListener.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2020 the original author or authors.
+ * Copyright 2015-2021 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.
@@ -40,7 +40,7 @@ public class LoggingProducerListener<K, V> implements ProducerListener<K, V> {
 	 */
 	public static final int DEFAULT_MAX_CONTENT_LOGGED = 100;
 
-	private static final LogAccessor LOGGER = new LogAccessor(LogFactory.getLog(LoggingProducerListener.class));
+	protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass())); // NOSONAR
 
 	private boolean includeContents = true;
 
@@ -68,15 +68,15 @@ public void setMaxContentLogged(int maxContentLogged) {
 
 	@Override
 	public void onError(ProducerRecord<K, V> record, @Nullable RecordMetadata recordMetadata, Exception exception) {
-		LOGGER.error(exception, () -> {
+		this.logger.error(exception, () -> {
 			StringBuffer logOutput = new StringBuffer();
 			logOutput.append("Exception thrown when sending a message");
 			if (this.includeContents) {
 				logOutput.append(" with key='")
-					.append(toDisplayString(ObjectUtils.nullSafeToString(record.key()), this.maxContentLogged))
+					.append(keyOrValue(record.key()))
 					.append("'")
 					.append(" and payload='")
-					.append(toDisplayString(ObjectUtils.nullSafeToString(record.value()), this.maxContentLogged))
+					.append(keyOrValue(record.value()))
 					.append("'");
 			}
 			logOutput.append(" to topic ").append(record.topic());
@@ -90,6 +90,15 @@ public void onError(ProducerRecord<K, V> record, @Nullable RecordMetadata record
 		});
 	}
 
+	private String keyOrValue(Object keyOrValue) {
+		if (keyOrValue instanceof byte[]) {
+			return "byte[" + ((byte[]) keyOrValue).length + "]";
+		}
+		else {
+			return toDisplayString(ObjectUtils.nullSafeToString(keyOrValue), this.maxContentLogged);
+		}
+	}
+
 	private String toDisplayString(String original, int maxCharacters) {
 		if (original.length() <= maxCharacters) {
 			return original;

spring-kafka/src/test/java/org/springframework/kafka/support/LoggingProducerListenerTests.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2021 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.support;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.BDDMockito.willAnswer;
+import static org.mockito.Mockito.spy;
+
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
+
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.core.log.LogAccessor;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+
+/**
+ * @author Gary Russell
+ * @since 2.5.16
+ *
+ */
+public class LoggingProducerListenerTests {
+
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	@Test
+	void noBytesInLog() {
+		LoggingProducerListener pl = new LoggingProducerListener();
+		LogAccessor logger = (LogAccessor) spy(KafkaTestUtils.getPropertyValue(pl, "logger"));
+		new DirectFieldAccessor(pl).setPropertyValue("logger", logger);
+		AtomicReference<String> string = new AtomicReference<>();
+		willAnswer(inv -> {
+			Supplier<String> stringer = inv.getArgument(1);
+			string.set(stringer.get());
+			return null;
+		}).given(logger).error(any(), any(Supplier.class));
+		pl.onError(new ProducerRecord("foo", 0, new byte[3], new byte[1111]), null,
+				new RuntimeException());
+		assertThat(string.get()).contains("byte[3]");
+		assertThat(string.get()).contains("byte[1111]");
+	}
+
+}