feature: Retry count available in header

README.adoc

@@ -474,7 +474,7 @@ You can also optionally provide a callback function to be run after the message(
 .Usage - print message content out to the console in parallel
 [source,java,indent=0]
         parallelConsumer.pollAndProduce(record -> {
-                    var result = processBrokerRecord(record);
+            var result = processBrokerRecord(record);
                     return new ProducerRecord<>(outputTopic, record.key(), result.payload);
                 }, consumeProduceResult -> {
                     log.debug("Message {} saved to broker at offset {}",
@@ -691,24 +691,21 @@ The system will treat this as a record processing success, mark the record as co
 
 A user may choose to skip a record for example, if it has been retried too many times or if the record is invalid or doesn't need processing.
 
+The number of failed attempts for a message can be retrevied through the API, see below for an example.
+
 Implementing a https://github.com/confluentinc/parallel-consumer/issues/196[max retries feature] as a part of the system is planned.
 
 .Example of skipping a record after a maximum number of retries is reached
 [source,java,indent=0]
 ----
         final int maxRetries = 10;
-        final Map<ConsumerRecord<String, String>, Long> retriesCount = new ConcurrentHashMap<>();
 
         pc.poll(consumerRecord -> {
-            Long retryCount = retriesCount.computeIfAbsent(consumerRecord, ignore -> 0L);
+            int retryCount = ParallelConsumerRecord.getFailedCount(consumerRecord);
             if (retryCount < maxRetries) {
                 processRecord(consumerRecord);
-                // no exception, so completed - remove from map
-                retriesCount.remove(consumerRecord);
             } else {
-                log.warn("Retry count {} exceeded max of {} for record {}", retryCount, maxRetries, consumerRecord);
-                // giving up, remove from map
-                retriesCount.remove(consumerRecord);
+                log.warn("Retry count {} exceeded max of {} for record, skipping. {}", retryCount, maxRetries, consumerRecord);
             }
         });
 ----

parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/ParallelConsumerRecord.java

@@ -0,0 +1,38 @@
+package io.confluent.parallelconsumer;
+
+/*-
+ * Copyright (C) 2020-2022 Confluent, Inc.
+ */
+
+import io.confluent.parallelconsumer.state.WorkContainer;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.Headers;
+
+import java.nio.ByteBuffer;
+
+/**
+ * Utility methods for Consumer Records
+ */
+public class ParallelConsumerRecord {
+
+    /**
+     * Get the number of failed attempts for a record.
+     *
+     * @param record the record
+     * @return the number of failed attempts
+     */
+    // TODO consider making this information available through a read only ConsumerRecord wrapper that only public API info (as WorkContainer is very internal and writeabe)
+    public static int getFailedCount(final ConsumerRecord<String, String> record) {
+        Headers headers = record.headers();
+        Header header = headers.lastHeader(WorkContainer.FAILED_COUNT_HEADER);
+        if (header != null) {
+            byte[] value = header.value();
+            ByteBuffer wrap = ByteBuffer.wrap(value);
+            return wrap.getInt();
+        } else {
+            return 0;
+        }
+    }
+
+}

parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/WorkContainer.java

@@ -1,7 +1,7 @@
 package io.confluent.parallelconsumer.state;
 
 /*-
- * Copyright (C) 2020-2021 Confluent, Inc.
+ * Copyright (C) 2020-2022 Confluent, Inc.
  */
 
 import io.confluent.csid.utils.WallClock;
@@ -13,7 +13,10 @@
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.header.internals.RecordHeader;
 
+import java.nio.ByteBuffer;
 import java.time.Duration;
 import java.time.Instant;
 import java.time.temporal.Temporal;
@@ -28,6 +31,7 @@
 @EqualsAndHashCode
 public class WorkContainer<K, V> implements Comparable<WorkContainer> {
 
+    public static final String FAILED_COUNT_HEADER = "X-Confluent-failed-count";
     private final String DEFAULT_TYPE = "DEFAULT";
 
     /**
@@ -89,11 +93,23 @@ public WorkContainer(int epoch, ConsumerRecord<K, V> cr, String workType) {
 
     public void fail(WallClock clock) {
         log.trace("Failing {}", this);
-        numberOfFailedAttempts++;
+        updateFailedCount();
         failedAt = Optional.of(clock.getNow());
         inFlight = false;
     }
 
+    private void updateFailedCount() {
+        numberOfFailedAttempts++;
+
+        byte[] encodedRetries = ByteBuffer.allocate(Integer.BYTES)
+                .putInt(getNumberOfFailedAttempts())
+                .array();
+        Headers headers = this.getCr().headers();
+        // replace
+        headers.remove(FAILED_COUNT_HEADER);
+        headers.add(new RecordHeader(FAILED_COUNT_HEADER, encodedRetries));
+    }
+
     public void succeed() {
         log.trace("Succeeded {}", this);
         inFlight = false;

parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/FailedCountTest.java

@@ -0,0 +1,44 @@
+package io.confluent.parallelconsumer;
+
+/*-
+ * Copyright (C) 2020-2022 Confluent, Inc.
+ */
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static com.google.common.truth.Truth.assertThat;
+
+class FailedCountTest extends ParallelEoSStreamProcessorTestBase {
+
+    @Test
+    void failedCountTest() {
+        primeFirstRecord();
+
+        AtomicInteger iteration = new AtomicInteger();
+        AtomicReference<ConsumerRecord> lastRec = new AtomicReference<>();
+        parallelConsumer.poll(consumerRecord -> {
+            iteration.incrementAndGet();
+
+            lastRec.set(consumerRecord);
+            int failedCount = ParallelConsumerRecord.getFailedCount(consumerRecord);
+
+            if (iteration.get() == 1) {
+                assertThat(failedCount).isEqualTo(0);
+                throw new RuntimeException("fail 1");
+            } else if (iteration.get() == 2) {
+                assertThat(failedCount).isEqualTo(1);
+            }
+        });
+        waitForOneLoopCycle();
+        parallelConsumer.closeDrainFirst();
+
+        ConsumerRecord consumerRecord = lastRec.get();
+        assertThat(consumerRecord).isNotNull();
+        int failedCount = ParallelConsumerRecord.getFailedCount(consumerRecord);
+        assertThat(failedCount).isEqualTo(1);
+    }
+}

parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/state/WorkManagerTest.java

@@ -1,9 +1,8 @@
 package io.confluent.parallelconsumer.state;
 
 /*-
- * Copyright (C) 2020-2021 Confluent, Inc.
+ * Copyright (C) 2020-2022 Confluent, Inc.
  */
-
 import io.confluent.csid.utils.AdvancingWallClockProvider;
 import io.confluent.csid.utils.KafkaTestUtils;
 import io.confluent.parallelconsumer.ParallelConsumerOptions;
@@ -245,7 +244,7 @@ void testOrderedAndDelayed() {
 
     @Test
     void containerDelay() {
-        var wc = new WorkContainer<String, String>(0, null);
+        var wc = new WorkContainer<String, String>(0, new ConsumerRecord<>("", 0, 0, "", ""));
         assertThat(wc.hasDelayPassed(clock)).isTrue(); // when new, there's no delay
         wc.fail(clock);
         assertThat(wc.hasDelayPassed(clock)).isFalse();

parallel-consumer-examples/parallel-consumer-example-core/src/main/java/io/confluent/parallelconsumer/examples/core/CoreApp.java

@@ -3,7 +3,9 @@
 /*-
  * Copyright (C) 2020-2022 Confluent, Inc.
  */
+
 import io.confluent.parallelconsumer.ParallelConsumerOptions;
+import io.confluent.parallelconsumer.ParallelConsumerRecord;
 import io.confluent.parallelconsumer.ParallelStreamProcessor;
 import lombok.Value;
 import lombok.extern.slf4j.Slf4j;
@@ -95,7 +97,7 @@ void runPollAndProduce() {
 
         // tag::exampleProduce[]
         parallelConsumer.pollAndProduce(record -> {
-                    var result = processBrokerRecord(record);
+            var result = processBrokerRecord(record);
                     return new ProducerRecord<>(outputTopic, record.key(), result.payload);
                 }, consumeProduceResult -> {
                     log.debug("Message {} saved to broker at offset {}",
@@ -134,18 +136,13 @@ void maxRetries() {
         ParallelStreamProcessor<String, String> pc = ParallelStreamProcessor.createEosStreamProcessor(null);
         // tag::maxRetries[]
         final int maxRetries = 10;
-        final Map<ConsumerRecord<String, String>, Long> retriesCount = new ConcurrentHashMap<>();
 
         pc.poll(consumerRecord -> {
-            Long retryCount = retriesCount.computeIfAbsent(consumerRecord, ignore -> 0L);
+            int retryCount = ParallelConsumerRecord.getFailedCount(consumerRecord);
             if (retryCount < maxRetries) {
                 processRecord(consumerRecord);
-                // no exception, so completed - remove from map
-                retriesCount.remove(consumerRecord);
             } else {
-                log.warn("Retry count {} exceeded max of {} for record {}", retryCount, maxRetries, consumerRecord);
-                // giving up, remove from map
-                retriesCount.remove(consumerRecord);
+                log.warn("Retry count {} exceeded max of {} for record, skipping. {}", retryCount, maxRetries, consumerRecord);
             }
         });
         // end::maxRetries[]

pom.xml

@@ -499,6 +499,7 @@
                 <artifactId>maven-surefire-plugin</artifactId>
                 <version>${surefire.version}</version>
                 <configuration>
+                    <trimStackTrace>false</trimStackTrace>
                     <excludes>
                         <!-- <exclude>**/*integration*/**/*.java</exclude> too aggressive? -->
                         <exclude>**/integrationTest*/**/*.java</exclude>

src/docs/README.adoc

@@ -604,6 +604,8 @@ The system will treat this as a record processing success, mark the record as co
 
 A user may choose to skip a record for example, if it has been retried too many times or if the record is invalid or doesn't need processing.
 
+The number of failed attempts for a message can be retrevied through the API, see below for an example.
+
 Implementing a https://github.com/confluentinc/parallel-consumer/issues/196[max retries feature] as a part of the system is planned.
 
 .Example of skipping a record after a maximum number of retries is reached