From b7cd6fe885a2bf278c5c9d34b97e8af4e9ebe56a Mon Sep 17 00:00:00 2001 From: Vishal Kumar Singh Date: Wed, 22 Apr 2026 11:37:14 +0530 Subject: [PATCH] fix: trim noisy user-function error log; move full PollContext to DEBUG The ERROR path in AbstractParallelEoSStreamProcessor#runUserFunction used to log the entire PollContextInternal (which is @ToString'd via Lombok and transitively prints every ConsumerRecord in the batch) on every user-function failure. With reasonable batch sizes this easily overflows the line length budget of log shippers (Filebeat, Fluent Bit, journald) and ends up truncated. It also leaks record payloads into error logs, which is awkward when those payloads contain sensitive data -- users were already working around this by masking the log pattern at log4j level. Keep a concise ERROR line (partitions + record count) and move the full PollContext dump to DEBUG so the signal stays useful without polluting error logs. The existing DEBUG-only PCRetriableException branch is preserved. Fixes #640 --- .../AbstractParallelEoSStreamProcessor.java | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/AbstractParallelEoSStreamProcessor.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/AbstractParallelEoSStreamProcessor.java index fc716650e..1f0bfc6e8 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/AbstractParallelEoSStreamProcessor.java +++ b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/AbstractParallelEoSStreamProcessor.java @@ -1351,12 +1351,23 @@ protected List, R>> runUserFunct } catch (Exception e) { // handle fail var cause = e.getCause(); + // Keep the primary error log line short (record count + partitions) so it stays readable + // when the user function throws for large batches; dumping the whole PollContext toString + // (which transitively prints every ConsumerRecord in the batch) tends to get truncated by + // downstream log tooling and can surface user-payload data inside error logs. The full + // context is still available at DEBUG for troubleshooting. + // See https://github.com/confluentinc/parallel-consumer/issues/640 + long recordCount = context.getPollContext().size(); + var partitions = context.getPollContext().getByTopicPartitionMap().keySet(); String msg = msg("Exception caught in user function running stage, registering WC as failed, returning to" + - " mailbox. Context: {}", context, e); + " mailbox. partitions={} records={}", partitions, recordCount); if (cause instanceof PCRetriableException) { log.debug("Explicit " + PCRetriableException.class.getSimpleName() + " caught, logging at DEBUG only. " + msg, e); } else { log.error(msg, e); + if (log.isDebugEnabled()) { + log.debug("Full PollContext for the failed user function batch: {}", context); + } } for (var wc : workContainerBatch) {