{"record":{"id":"4885af60bfd71ff3","repo":"apache/beam","slug":"outbound-data-endpoint-already-registered-for","errorCode":null,"errorMessage":"Outbound data endpoint already registered for ","messagePattern":"Outbound data endpoint already registered for ","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataOutboundAggregator.java","lineNumber":143,"sourceCode":"  public void start() {\n    if (timeLimit > 0 && this.flushFuture == null) {\n      this.flushFuture =\n          Executors.newSingleThreadScheduledExecutor(\n                  new ThreadFactoryBuilder()\n                      .setDaemon(true)\n                      .setNameFormat(\"DataBufferOutboundFlusher-thread\")\n                      .build())\n              .scheduleAtFixedRate(this::flush, timeLimit, timeLimit, TimeUnit.MILLISECONDS);\n    }\n  }\n\n  /**\n   * Register the outbound data logical endpoint, returns the FnDataReceiver for processing the\n   * endpoint's outbound data.\n   */\n  public <T> FnDataReceiver<T> registerOutputDataLocation(String pTransformId, Coder<T> coder) {\n    if (outputDataReceivers.containsKey(pTransformId)) {\n      throw new IllegalStateException(\n          \"Outbound data endpoint already registered for \" + pTransformId);\n    }\n    Receiver<T> receiver = new Receiver<>(coder);\n    if (timeLimit > 0) {\n      outputDataReceivers.put(pTransformId, receiver);\n      return data -> {\n        checkFlushThreadException();\n        synchronized (flushLock) {\n          receiver.accept(data);\n        }\n      };\n    }\n    outputDataReceivers.put(pTransformId, receiver);\n    return receiver;\n  }\n\n  /**\n   * Register the outbound timers logical endpoint, returns the FnDataReceiver for processing the","sourceCodeStart":125,"sourceCodeEnd":161,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataOutboundAggregator.java#L125-L161","documentation":"BeamFnDataOutboundAggregator registers one outbound logical data endpoint per pTransformId per bundle. This IllegalStateException is thrown when registerOutputDataLocation is called with a pTransformId that already has a registered data receiver, preventing duplicate endpoint registration which would corrupt the bundle's output mapping.","triggerScenarios":"Calling registerOutputDataLocation twice with the same pTransformId within one bundle; transform/receiver wiring (e.g. via addOutgoingDataEndpoint or fnDataReceiver) registering the same output location twice.","commonSituations":"Runner/SDK harness bugs where bundle setup runs the registration path twice (a PTransform re-initialized mid-bundle); custom runner code calling registerOutputLocation without clearing prior state between bundles.","solutions":["Ensure registerOutputDataLocation is called at most once per pTransformId per bundle; check existing registrations before calling","Check that the wiring code (addOutgoingDataEndpoint, registerOutputLocation, newBundle) is not invoked redundantly","Upgrade Beam — bundle lifecycle bugs in custom runners have been fixed in later versions","Recreate/reset the aggregator between bundles so stale registrations are cleared"],"exampleFix":"// before\naggregator.registerOutputDataLocation(pTransformId, coder);\naggregator.registerOutputDataLocation(pTransformId, coder); // throws\n// after\n// register exactly once per pTransformId per bundle","handlingStrategy":"validation","validationCode":"if (registeredTransformIds.contains(pTransformId)) {\n  throw new IllegalArgumentException(\"duplicate data endpoint \" + pTransformId);\n}\nregisteredTransformIds.add(pTransformId);","typeGuard":null,"tryCatchPattern":"try {\n  aggregator.registerOutputDataLocation(pTransformId, coder);\n} catch (IllegalStateException e) {\n  LOG.warn(\"endpoint already registered: \" + pTransformId, e);\n}","preventionTips":["Register each pTransformId exactly once per bundle","Centralize endpoint registration in newBundle/setup, not per-element code","Track registered ids in a Set during bundle setup"],"tags":["java","beam","state","duplicate-registration"],"backgroundTag":"invalid-state-transition","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}