Skip to content

Commit 297f52d

Browse files
garyrussellartembilan
garyrussell
authored and
artembilan
committed
GH-2132: Producer Record Log Formatting
Partial solution for #2132 Allow the user to provide a function for formatting producer records. Also move consumer record formatting to `KafkaUtils` will deprecate `ListenerUtils` versions later. There will be more polising in this area for 3.0 only (separate PR). **cherry-pick to 2.8.x, 2.7.x**
1 parent 8291609 commit 297f52d

File tree

5 files changed

+112
-28
lines changed

5 files changed

+112
-28
lines changed

‎spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -612,7 +612,7 @@ protected void closeProducer(Producer<K, V> producer, boolean inTx) {
612612
*/
613613
protected ListenableFuture<SendResult<K, V>> doSend(final ProducerRecord<K, V> producerRecord) {
614614
final Producer<K, V> producer = getTheProducer(producerRecord.topic());
615-
this.logger.trace(() -> "Sending: " + producerRecord);
615+
this.logger.trace(() -> "Sending: " + KafkaUtils.format(producerRecord));
616616
final SettableListenableFuture<SendResult<K, V>> future = new SettableListenableFuture<>();
617617
Object sample = null;
618618
if (this.micrometerEnabled && this.micrometerHolder == null) {
@@ -639,7 +639,7 @@ protected ListenableFuture<SendResult<K, V>> doSend(final ProducerRecord<K, V> p
639639
if (this.autoFlush) {
640640
flush();
641641
}
642-
this.logger.trace(() -> "Sent: " + producerRecord);
642+
this.logger.trace(() -> "Sent: " + KafkaUtils.format(producerRecord));
643643
return future;
644644
}
645645

@@ -656,7 +656,8 @@ private Callback buildCallback(final ProducerRecord<K, V> producerRecord, final
656656
if (KafkaTemplate.this.producerListener != null) {
657657
KafkaTemplate.this.producerListener.onSuccess(producerRecord, metadata);
658658
}
659-
KafkaTemplate.this.logger.trace(() -> "Sent ok: " + producerRecord + ", metadata: " + metadata);
659+
KafkaTemplate.this.logger.trace(() -> "Sent ok: " + KafkaUtils.format(producerRecord)
660+
+ ", metadata: " + metadata);
660661
}
661662
else {
662663
if (sample != null) {
@@ -666,7 +667,8 @@ private Callback buildCallback(final ProducerRecord<K, V> producerRecord, final
666667
if (KafkaTemplate.this.producerListener != null) {
667668
KafkaTemplate.this.producerListener.onError(producerRecord, metadata, exception);
668669
}
669-
KafkaTemplate.this.logger.debug(exception, () -> "Failed to send: " + producerRecord);
670+
KafkaTemplate.this.logger.debug(exception, () -> "Failed to send: "
671+
+ KafkaUtils.format(producerRecord));
670672
}
671673
}
672674
finally {

‎spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerUtils.java

Lines changed: 4 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@
2929
import org.apache.kafka.common.header.internals.RecordHeaders;
3030

3131
import org.springframework.core.log.LogAccessor;
32+
import org.springframework.kafka.support.KafkaUtils;
3233
import org.springframework.kafka.support.serializer.DeserializationException;
3334
import org.springframework.lang.Nullable;
3435
import org.springframework.util.Assert;
@@ -47,8 +48,6 @@ public final class ListenerUtils {
4748
private ListenerUtils() {
4849
}
4950

50-
private static final ThreadLocal<Boolean> LOG_METADATA_ONLY = new ThreadLocal<>();
51-
5251
private static final int DEFAULT_SLEEP_INTERVAL = 100;
5352

5453
private static final int SMALL_SLEEP_INTERVAL = 10;
@@ -121,7 +120,7 @@ public static DeserializationException getExceptionFromHeader(final ConsumerReco
121120
* @see #recordToString(ConsumerRecord)
122121
*/
123122
public static void setLogOnlyMetadata(boolean onlyMeta) {
124-
LOG_METADATA_ONLY.set(onlyMeta);
123+
KafkaUtils.setLogOnlyMetadata(onlyMeta);
125124
}
126125

127126
/**
@@ -133,12 +132,7 @@ public static void setLogOnlyMetadata(boolean onlyMeta) {
133132
* @see #setLogOnlyMetadata(boolean)
134133
*/
135134
public static String recordToString(ConsumerRecord<?, ?> record) {
136-
if (Boolean.TRUE.equals(LOG_METADATA_ONLY.get())) {
137-
return record.topic() + "-" + record.partition() + "@" + record.offset();
138-
}
139-
else {
140-
return record.toString();
141-
}
135+
return KafkaUtils.recordToString(record);
142136
}
143137

144138
/**
@@ -150,12 +144,7 @@ public static String recordToString(ConsumerRecord<?, ?> record) {
150144
* @since 2.5.4
151145
*/
152146
public static String recordToString(ConsumerRecord<?, ?> record, boolean meta) {
153-
if (meta) {
154-
return record.topic() + "-" + record.partition() + "@" + record.offset();
155-
}
156-
else {
157-
return record.toString();
158-
}
147+
return KafkaUtils.recordToString(record, meta);
159148
}
160149

161150
/**

‎spring-kafka/src/main/java/org/springframework/kafka/requestreply/AggregatingReplyingKafkaTemplate.java

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2019-2020 the original author or authors.
2+
* Copyright 2019-2022 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -39,6 +39,7 @@
3939
import org.springframework.kafka.listener.ContainerProperties.AckMode;
4040
import org.springframework.kafka.listener.GenericMessageListenerContainer;
4141
import org.springframework.kafka.support.KafkaHeaders;
42+
import org.springframework.kafka.support.KafkaUtils;
4243
import org.springframework.util.Assert;
4344

4445
/**
@@ -126,7 +127,7 @@ public void onMessage(List<ConsumerRecord<K, Collection<ConsumerRecord<K, R>>>>
126127
data.forEach(record -> {
127128
Header correlation = record.headers().lastHeader(KafkaHeaders.CORRELATION_ID);
128129
if (correlation == null) {
129-
this.logger.error(() -> "No correlationId found in reply: " + record
130+
this.logger.error(() -> "No correlationId found in reply: " + KafkaUtils.recordToString(record)
130131
+ " - to use request/reply semantics, the responding server must return the correlation id "
131132
+ " in the '" + KafkaHeaders.CORRELATION_ID + "' header");
132133
}

‎spring-kafka/src/main/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplate.java

Lines changed: 9 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2018-2021 the original author or authors.
2+
* Copyright 2018-2022 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -47,6 +47,7 @@
4747
import org.springframework.kafka.listener.GenericMessageListenerContainer;
4848
import org.springframework.kafka.listener.ListenerUtils;
4949
import org.springframework.kafka.support.KafkaHeaders;
50+
import org.springframework.kafka.support.KafkaUtils;
5051
import org.springframework.kafka.support.TopicPartitionOffset;
5152
import org.springframework.kafka.support.serializer.DeserializationException;
5253
import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
@@ -378,7 +379,7 @@ public RequestReplyFuture<K, V, R> sendAndReceive(ProducerRecord<K, V> record, @
378379
}
379380
}
380381
headers.add(new RecordHeader(this.correlationHeaderName, correlationId.getCorrelationId()));
381-
this.logger.debug(() -> "Sending: " + record + WITH_CORRELATION_ID + correlationId);
382+
this.logger.debug(() -> "Sending: " + KafkaUtils.format(record) + WITH_CORRELATION_ID + correlationId);
382383
RequestReplyFuture<K, V, R> future = new RequestReplyFuture<>();
383384
this.futures.put(correlationId, future);
384385
try {
@@ -396,7 +397,8 @@ private void scheduleTimeout(ProducerRecord<K, V> record, CorrelationKey correla
396397
this.scheduler.schedule(() -> {
397398
RequestReplyFuture<K, V, R> removed = this.futures.remove(correlationId);
398399
if (removed != null) {
399-
this.logger.warn(() -> "Reply timed out for: " + record + WITH_CORRELATION_ID + correlationId);
400+
this.logger.warn(() -> "Reply timed out for: " + KafkaUtils.format(record)
401+
+ WITH_CORRELATION_ID + correlationId);
400402
if (!handleTimeout(correlationId, removed)) {
401403
removed.setException(new KafkaReplyTimeoutException("Reply timed out"));
402404
}
@@ -455,7 +457,7 @@ public void onMessage(List<ConsumerRecord<K, R>> data) {
455457
correlationId = new CorrelationKey(correlationHeader.value());
456458
}
457459
if (correlationId == null) {
458-
this.logger.error(() -> "No correlationId found in reply: " + record
460+
this.logger.error(() -> "No correlationId found in reply: " + KafkaUtils.recordToString(record)
459461
+ " - to use request/reply semantics, the responding server must return the correlation id "
460462
+ " in the '" + this.correlationHeaderName + "' header");
461463
}
@@ -473,7 +475,8 @@ public void onMessage(List<ConsumerRecord<K, R>> data) {
473475
future.setException(exception);
474476
}
475477
if (ok) {
476-
this.logger.debug(() -> "Received: " + record + WITH_CORRELATION_ID + correlationKey);
478+
this.logger.debug(() -> "Received: " + KafkaUtils.recordToString(record)
479+
+ WITH_CORRELATION_ID + correlationKey);
477480
future.set(record);
478481
}
479482
}
@@ -540,7 +543,7 @@ protected void logLateArrival(ConsumerRecord<K, R> record, CorrelationKey correl
540543
}
541544

542545
private String missingCorrelationLogMessage(ConsumerRecord<K, R> record, CorrelationKey correlationId) {
543-
return "No pending reply: " + record + WITH_CORRELATION_ID
546+
return "No pending reply: " + KafkaUtils.recordToString(record) + WITH_CORRELATION_ID
544547
+ correlationId + ", perhaps timed out, or using a shared reply topic";
545548
}
546549

‎spring-kafka/src/main/java/org/springframework/kafka/support/KafkaUtils.java

Lines changed: 90 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2018-2021 the original author or authors.
2+
* Copyright 2018-2022 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -22,10 +22,14 @@
2222
import java.time.Duration;
2323
import java.util.Collection;
2424
import java.util.Map;
25+
import java.util.function.Function;
2526

27+
import org.apache.kafka.clients.consumer.ConsumerRecord;
2628
import org.apache.kafka.clients.producer.ProducerConfig;
29+
import org.apache.kafka.clients.producer.ProducerRecord;
2730

2831
import org.springframework.messaging.Message;
32+
import org.springframework.util.Assert;
2933
import org.springframework.util.ClassUtils;
3034

3135
/**
@@ -38,6 +42,13 @@
3842
*/
3943
public final class KafkaUtils {
4044

45+
private static final ThreadLocal<Boolean> LOG_METADATA_ONLY = new ThreadLocal<>();
46+
47+
private static Function<ProducerRecord<?, ?>, String> prFormatter = rec -> rec.toString();
48+
49+
private static Function<ConsumerRecord<?, ?>, String> crFormatter =
50+
rec -> rec.topic() + "-" + rec.partition() + "@" + rec.offset();
51+
4152
/**
4253
* True if micrometer is on the class path.
4354
*/
@@ -134,6 +145,84 @@ else if (dt instanceof String) {
134145
min));
135146
}
136147

148+
/**
149+
* Set to true to only log record metadata.
150+
* @param onlyMeta true to only log record metadata.
151+
* @since 2.7.12
152+
* @see #recordToString(ConsumerRecord)
153+
*/
154+
public static void setLogOnlyMetadata(boolean onlyMeta) {
155+
LOG_METADATA_ONLY.set(onlyMeta);
156+
}
157+
158+
/**
159+
* Return the {@link ConsumerRecord} as a String; either {@code toString()} or
160+
* {@code topic-partition@offset}.
161+
* @param record the record.
162+
* @return the rendered record.
163+
* @since 2.7.12
164+
* @see #setLogOnlyMetadata(boolean)
165+
*/
166+
public static String recordToString(ConsumerRecord<?, ?> record) {
167+
return recordToString(record, Boolean.TRUE.equals(LOG_METADATA_ONLY.get()));
168+
}
169+
170+
/**
171+
* Return the {@link ConsumerRecord} as a String; either {@code toString()} or
172+
* {@code topic-partition@offset}.
173+
* @param record the record.
174+
* @param meta true to log just the metadata.
175+
* @return the rendered record.
176+
* @since 2.7.12
177+
*/
178+
public static String recordToString(ConsumerRecord<?, ?> record, boolean meta) {
179+
if (meta) {
180+
return crFormatter.apply(record);
181+
}
182+
else {
183+
return record.toString();
184+
}
185+
}
186+
187+
/**
188+
* Set a formatter for logging {@link ConsumererRecord}s.
189+
* @param formatter a function to format the record as a String
190+
* @since 2.7.11
191+
*/
192+
public static void setConsumerRecordFormatter(Function<ConsumerRecord<?, ?>, String> formatter) {
193+
Assert.notNull(formatter, "'formatter' cannot be null");
194+
crFormatter = formatter;
195+
}
196+
197+
/**
198+
* Set a formatter for logging {@link ProducerRecord}s.
199+
* @param formatter a function to format the record as a String
200+
* @since 2.7.11
201+
*/
202+
public static void setProducerRecordFormatter(Function<ProducerRecord<?, ?>, String> formatter) {
203+
Assert.notNull(formatter, "'formatter' cannot be null");
204+
prFormatter = formatter;
205+
}
206+
207+
/**
208+
* Format the {@link ConsumerRecord} for logging; default
209+
* {@code topic-partition@offset}.
210+
* @param record the record to format.
211+
* @return the formatted String.
212+
*/
213+
public static String format(ConsumerRecord<?, ?> record) {
214+
return crFormatter.apply(record);
215+
}
216+
217+
/**
218+
* Format the {@link ProducerRecord} for logging; default
219+
* {@link ProducerRecord}{@link #toString()}.
220+
* @param record the record to format.
221+
* @return the formatted String.
222+
*/
223+
public static String format(ProducerRecord<?, ?> record) {
224+
return prFormatter.apply(record);
225+
}
137226

138227
private KafkaUtils() {
139228
}

0 commit comments

Comments
 (0)