show record topic in slow-work warning message

confluentinc · Jul 27, 2021 · faa40c2 · faa40c2
1 parent 7458405
commit faa40c2
Showing 1 changed file with 4 additions and 3 deletions.
diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/WorkManager.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/WorkManager.java
@@ -246,6 +246,7 @@ public List<WorkContainer<K, V>> maybeGetWork(int requestedMaxWorkToRetrieve) {
         var staleWorkToRemove = new ArrayList<WorkContainer<K, V>>();
 
         var slowWorkCount = 0;
+        var slowWorkTopics = new HashSet<String>();
 
         //
         for (var shard : it) {
@@ -306,6 +307,7 @@ public List<WorkContainer<K, V>> maybeGetWork(int requestedMaxWorkToRetrieve) {
                     String msg = "Can't take as work: Work ({}). Must all be true: Delay passed= {}. Is not in flight= {}. Has not succeeded already= {}. Time spent in execution queue: {}.";
                     if (toSeconds(timeInFlight) > toSeconds(thresholdForTimeSpentInQueueWarning)) {
                         slowWorkCount++;
+                        slowWorkTopics.add(workContainer.getCr().topic());
                         log.trace("Work has spent over " + thresholdForTimeSpentInQueueWarning + " in queue! "
                                 + msg, workContainer, delayHasPassed, workContainer.isNotInFlight(), hasNotSucceededAlready, timeInFlight);
                     } else {
@@ -330,9 +332,8 @@ public List<WorkContainer<K, V>> maybeGetWork(int requestedMaxWorkToRetrieve) {
         if (slowWorkCount > 0) {
             final int finalSlowWorkCount = slowWorkCount;
             slowWarningRateLimit.performIfNotLimited(() -> log.warn("Warning: {} records in the queue have been " +
-                            "waiting longer than {}.",
-                    finalSlowWorkCount, toSeconds(thresholdForTimeSpentInQueueWarning)));
-
+                            "waiting longer than {}s for following topics {}.",
+                    finalSlowWorkCount, toSeconds(thresholdForTimeSpentInQueueWarning), slowWorkTopics));
         }
 
         // remove found stale work outside of loop