log improvements

confluentinc · Dec 29, 2022 · da9f879 · da9f879
1 parent 908df5d
commit da9f879
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 2 deletions.
diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/PartitionState.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/PartitionState.java
@@ -484,7 +484,10 @@ private boolean updateBlockFromEncodingResult(String offsetMapPayload) {
                     getTp(), metaPayloadLength, getPressureThresholdValue(), DefaultMaxMetadataSize);
 
         } else { // and thus (metaPayloadLength <= pressureThresholdValue)
-            setAllowedMoreRecords(true);
+            if (allowedMoreRecords == false) {
+                // guard is useful for debugging to catch the transition from false to true
+                setAllowedMoreRecords(true);
+            }
             log.debug("Payload size {} within threshold {}", metaPayloadLength, getPressureThresholdValue());
         }
 
@@ -591,9 +594,11 @@ public boolean couldBeTakenAsWork(WorkContainer<K, V> workContainer) {
             log.debug("Work is in queue with stale epoch or no longer assigned. Skipping. Shard it came from will/was removed during partition revocation. WC: {}", workContainer);
             return false;
         } else if (isAllowedMoreRecords()) {
+            log.debug("Partition is allowed more records. Taking work. WC: {}", workContainer);
             return true;
         } else if (isBlockingProgress(workContainer)) {
             // allow record to be taken, even if partition is blocked, as this record completion may reduce payload size requirement
+            log.debug("Partition is blocked, but this record is blocking progress. Taking work. WC: {}", workContainer);
             return true;
         } else {
             log.debug("Not allowed more records for the partition ({}) as set from previous encode run (blocked), that this " +

diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/WorkContainer.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/WorkContainer.java
@@ -201,7 +201,7 @@ public boolean isUserFunctionSucceeded() {
 
     @Override
     public String toString() {
-        return "WorkContainer(" + toTopicPartition(cr) + ":" + cr.offset() + ":" + cr.key() + ")";
+        return "WorkContainer(tp:" + toTopicPartition(cr) + ":o:" + cr.offset() + ":k:" + cr.key() + ")";
     }
 
     public Duration getTimeInFlight() {