{"record":{"id":"275baea6f93386f3","repo":"apache/beam","slug":"outbound-timers-endpoint-already-registered-for-timerkey","errorCode":null,"errorMessage":"Outbound timers endpoint already registered for ${timerKey}","messagePattern":"Outbound timers 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":168,"sourceCode":"        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\n   * endpoint's outbound timers data.\n   */\n  public <T> FnDataReceiver<T> registerOutputTimersLocation(\n      String pTransformId, String timerFamilyId, Coder<T> coder) {\n    TimerEndpoint timerKey = new TimerEndpoint(pTransformId, timerFamilyId);\n    if (outputTimersReceivers.containsKey(timerKey)) {\n      throw new IllegalStateException(\n          \"Outbound timers endpoint already registered for \" + timerKey);\n    }\n    Receiver<T> receiver = new Receiver<>(coder);\n    if (timeLimit > 0) {\n      outputTimersReceivers.put(timerKey, receiver);\n      return timers -> {\n        checkFlushThreadException();\n        synchronized (flushLock) {\n          receiver.accept(timers);\n        }\n      };\n    }\n    outputTimersReceivers.put(timerKey, receiver);\n    return receiver;\n  }\n\n  private void flushInternal() {\n    if (bytesWrittenSinceFlush == 0) {","sourceCodeStart":150,"sourceCodeEnd":186,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataOutboundAggregator.java#L150-L186","documentation":"Analogous to the data endpoint check: registerOutputTimersLocation registers one outbound timers receiver per (pTransformId, timerFamilyId) pair. This IllegalStateException is thrown when the same TimerEndpoint is registered twice, which would mean two receivers competing for the same timers channel.","triggerScenarios":"Calling registerOutputTimersLocation twice with identical pTransformId and timerFamilyId; duplicate wiring through addOutgoingTimersEndpoint or registerOutputLocation for the same timer family.","commonSituations":"Stateful DoFns with multiple timer families whose registration code runs twice per bundle; runner harness re-initialization without resetting outputTimersReceivers between bundles.","solutions":["Ensure each (pTransformId, timerFamilyId) pair is registered only once per bundle","Verify the timer family id is distinct where multiple timers per transform are intended","Check bundle setup code for double invocation of addOutgoingTimersEndpoint","Recreate the aggregator between bundles so stale registrations are cleared"],"exampleFix":"// before\naggregator.registerOutputTimersLocation(pTransformId, timerFamilyId, coder);\naggregator.registerOutputTimersLocation(pTransformId, timerFamilyId, coder); // throws\n// after\naggregator.registerOutputTimersLocation(pTransformId, timerFamilyId + \"-v2\", coder); // distinct family id","handlingStrategy":"validation","validationCode":"if (registeredTimerKeys.contains(pTransformId + \"/\" + timerFamilyId)) {\n  return; // skip duplicate registration\n}\nregisteredTimerKeys.add(pTransformId + \"/\" + timerFamilyId);","typeGuard":null,"tryCatchPattern":"try {\n  aggregator.registerOutputTimersLocation(pTransformId, timerFamilyId, coder);\n} catch (IllegalStateException e) {\n  LOG.warn(\"timer endpoint already registered: \" + timerFamilyId, e);\n}","preventionTips":["Use a unique timerFamilyId per timer declaration","Register timers only in one-time bundle setup","Keep a Set of registered TimerEndpoint keys"],"tags":["java","beam","timers","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"}