apache/beam · warning

Was unable to reset bundle processor safely. Bundle…

Error message

Was unable to reset bundle processor safely. Bundle processor will be discarded and re-instantiated on next bundle for descriptor {}.

What it means

ProcessBundleHandler.release() calls BundleProcessor.reset() when returning a processor to the cache after a bundle; if reset() throws, this warning is logged and the processor is discarded instead of cached. The next bundle for that descriptor will pay the cost of re-instantiating the processor. Any exception thrown by DoFn StartBundle/FinishBundle cleanup paths or registered reset handlers surfaces here.

Solutions

1. Fix the underlying exception included in the warning's stack trace (usually a failing close/finish callback).
2. Ensure DoFn external resources are closed idempotently and safely in FinishBundle/teardown.
3. Check for null or corrupted state after failed bundles before reset.
4. Investigate whether reset failures repeat for every bundle (then performance degrades — re-instantiation each time).

Example fix

// before
@FinishBundle
public void finish(FinishContext ctx) {
  writer.flush(); // throws if writer already closed
}
// after
@FinishBundle
public void finish(FinishContext ctx) {
  if (writer != null) {
    try { writer.flush(); } catch (IOException ignored) {}
  }
}

// Validate closable resources are open before reset in FinishBundle
if (writer == null || !writer.isOpen()) { LOG.warn("writer already closed; skip flush"); }

try {
  bundleProcessor.reset();
} catch (Exception e) {
  LOG.warn("reset failed; processor will be recreated", e);
}

When it happens

Trigger: bundleProcessor.reset() throws during release() after a bundle completes, typically because a DoFn's FinishBundle/setup-related state cleanup or a registered reset function raised an exception.

Common situations: DoFns holding external resources (connections, files) that fail to close cleanly; exceptions in FinishBundle; state corruption after a failed bundle.

Related errors

• Exception destroying pooled object.
• A function must be provided to convert the input type into…
• A PValue contained in
• A schema was provided without a data format (or viceversa)…
• All inherited interfaces of

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/bd13a280aa482788. Report an issue: GitHub.

Appendix: source

Thrown at sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/ProcessBundleHandler.java:1025

    /**
     * Finds an active bundle processor for the specified {@code instructionId} or null if one could
     * not be found.
     */
    public BundleProcessor find(String instructionId) {
      return activeBundleProcessors.get(instructionId);
    }

    /**
     * Add a {@link BundleProcessor} to cache. The {@link BundleProcessor} will be marked as
     * inactive and reset before being added to the cache.
     */
    void release(String bundleDescriptorId, BundleProcessor bundleProcessor) {
      activeBundleProcessors.remove(bundleProcessor.getInstructionId());
      try {
        bundleProcessor.reset();
        cachedBundleProcessors.get(bundleDescriptorId).add(bundleProcessor);
      } catch (Exception e) {
        LOG.warn(
            "Was unable to reset bundle processor safely. Bundle processor will be discarded and re-instantiated on next bundle for descriptor {}.",
            bundleDescriptorId,
            e);
      }
    }

    /** Discard an active {@link BundleProcessor} instead of being re-used. */
    void discard(BundleProcessor bundleProcessor) {
      bundleProcessor.discard();
      activeBundleProcessors.remove(bundleProcessor.getInstructionId());
    }

    /** Shutdown all the cached {@link BundleProcessor}s, running the tearDown() functions. */
    void shutdown() throws Exception {
      cachedBundleProcessors.invalidateAll();
    }
  }

View on GitHub (pinned to 12126d8942)