Improve tests when errors are received in the consumer thread (apache…

…#11858)
snleee · Oct 23, 2023 · 25b1fcb · 25b1fcb
1 parent 8c87710
commit 25b1fcb
Showing 1 changed file with 7 additions and 4 deletions.
diff --git a/...src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java b/...src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
@@ -391,12 +391,12 @@ private void handleTransientStreamErrors(Exception e)
     _serverMetrics.addMeteredTableValue(_tableStreamName, ServerMeter.REALTIME_CONSUMPTION_EXCEPTIONS,
         1L);
     if (_consecutiveErrorCount > MAX_CONSECUTIVE_ERROR_COUNT) {
-      _segmentLogger.warn("Stream transient exception when fetching messages, stopping consumption after {} attempts",
-          _consecutiveErrorCount, e);
+      _segmentLogger.warn("Stream transient exception when fetching messages, stopping consumption after "
+          + _consecutiveErrorCount + " attempts", e);
       throw e;
     } else {
-      _segmentLogger
-          .warn("Stream transient exception when fetching messages, retrying (count={})", _consecutiveErrorCount, e);
+      _segmentLogger.warn("Stream transient exception when fetching messages, retrying (count="
+          + _consecutiveErrorCount + ")", e);
       Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS);
       recreateStreamConsumer("Too many transient errors");
     }
@@ -444,6 +444,9 @@ protected boolean consumeLoop()
         // One such exception seen so far is java.net.SocketTimeoutException
         handleTransientStreamErrors(e);
         continue;
+      } catch (Throwable t) {
+        _segmentLogger.warn("Stream error when fetching messages, stopping consumption", t);
+        throw t;
       }
 
       boolean endCriteriaReached = processStreamEvents(messageBatch, idlePipeSleepTimeMillis);