{"record":{"id":"56b10f8594b99095","repo":"apache/beam","slug":"periodic-flushing-thread-finished-unexpectedly","errorCode":null,"errorMessage":"Periodic flushing thread finished unexpectedly.","messagePattern":"Periodic flushing thread finished unexpectedly\\.","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataOutboundAggregator.java","lineNumber":334,"sourceCode":"\n  void flush() {\n    try {\n      synchronized (flushLock) {\n        flushInternal();\n      }\n    } catch (OutOfMemoryError oom) {\n      throw oom;\n    } catch (Throwable t) {\n      throw new RuntimeException(t);\n    }\n  }\n\n  /** Check if the flush thread failed with an exception. */\n  private void checkFlushThreadException() throws IOException {\n    if (flushFuture != null && flushFuture.isDone()) {\n      try {\n        flushFuture.get();\n        throw new IOException(\"Periodic flushing thread finished unexpectedly.\");\n      } catch (ExecutionException ee) {\n        unwrapExecutionException(ee);\n      } catch (CancellationException ce) {\n        throw new IOException(ce);\n      } catch (InterruptedException ie) {\n        Thread.currentThread().interrupt();\n        throw new IOException(ie);\n      }\n    }\n  }\n\n  private void unwrapExecutionException(ExecutionException ee) throws IOException {\n    // the cause is always RuntimeException\n    RuntimeException re = (RuntimeException) ee.getCause();\n    if (re.getCause() instanceof IOException) {\n      throw (IOException) re.getCause();\n    } else {\n      throw new IOException(re.getCause());","sourceCodeStart":316,"sourceCodeEnd":352,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataOutboundAggregator.java#L316-L352","documentation":"The aggregator schedules a periodic flush task that drains buffered outbound data. checkFlushThreadException verifies the task is still running; if flushFuture is done but get() returns normally (no result), the thread terminated silently, which would leave data unflushed, so an IOException is thrown.","triggerScenarios":"The scheduled flush task completed without an exception while the aggregator is still active; detected on the next registerOutputDataLocation or registerOutputTimersLocation call. If the task failed with an exception, ExecutionException is unwrapped instead.","commonSituations":"Executor shutdown or misconfiguration causing the periodic task to terminate cleanly; a runner closing the aggregator's scheduler early while registration is still happening.","solutions":["Inspect the executor/scheduler used to schedule the flush task — ensure it is not shut down prematurely","Check logs preceding this error for scheduling or cancellation causes","Ensure the aggregator is only used while its bundle and flush thread are active","Upgrade Beam version; scheduling lifecycle bugs have been patched historically"],"exampleFix":"// before\nScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();\nexecutor.shutdown(); // flush thread finishes unexpectedly\n// after\n// shut down the executor only after aggregator.close()","handlingStrategy":"try-catch","validationCode":"if (flushFuture != null && flushFuture.isDone()) {\n  recreateAggregator();\n}","typeGuard":null,"tryCatchPattern":"try {\n  aggregator.registerOutputDataLocation(id, coder);\n} catch (IOException e) {\n  if (e.getMessage().contains(\"Periodic flushing\")) {\n    LOG.error(\"flush thread died unexpectedly\", e);\n  }\n  throw e;\n}","preventionTips":["Keep the scheduler executor alive for the aggregator's lifetime","Monitor flush task failures promptly via logs","Close the aggregator explicitly when done to stop the flush thread cleanly"],"tags":["java","beam","flush","threading"],"backgroundTag":"internal-invariant-violation","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}