Add exception handling for timeout

Signed-off-by: Chase Engelbrecht <[email protected]>
engechas · Mar 8, 2024 · d07feb6 · d07feb6
1 parent de6709d
commit d07feb6
Showing 1 changed file with 6 additions and 2 deletions.
diff --git a/...ffer-common/src/main/java/org/opensearch/dataprepper/buffer/common/BufferAccumulator.java b/...ffer-common/src/main/java/org/opensearch/dataprepper/buffer/common/BufferAccumulator.java
@@ -104,6 +104,10 @@ private boolean flushWithBackoff() throws Exception{
                 LOG.warn("Retrying of flushing the buffer accumulator was interrupted: {}", e.getMessage());
                 scheduledExecutorService.shutdownNow();
                 throw e;
+            } catch (final TimeoutException e) {
+                LOG.warn("Retrying of flushing the buffer accumulator did not complete in a reasonable duration: {}", e.getMessage());
+                scheduledExecutorService.shutdownNow();
+                throw e;
             }
         }
 
@@ -122,8 +126,8 @@ private void flushAccumulatedToBuffer() throws Exception {
     }
 
     private long getMaxFlushDuration(final long delay) {
-        // Max duration a flush should take + 5 seconds of buffer
-        return bufferTimeoutMillis + delay + 5000;
+        // Max duration a flush should take * 2
+        return (bufferTimeoutMillis + delay) * 2;
     }
 
     /**