{"record":{"id":"5e0c295634be100f","repo":"apache/druid","slug":"already-started-5e0c29","errorCode":null,"errorMessage":"Already started","messagePattern":"Already started","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"server/src/main/java/org/apache/druid/messages/client/MessageRelay.java","lineNumber":109,"sourceCode":"  /**\n   * Retrieves messages that are being sent to this client and hands them to {@link #listener}.\n   */\n  private class Collector\n  {\n    private final MessageListener<MessageType> listener;\n    private final AtomicLong epoch = new AtomicLong(INIT);\n    private final AtomicLong watermark = new AtomicLong(INIT);\n    private final AtomicReference<ListenableFuture<?>> currentCall = new AtomicReference<>();\n\n    public Collector(final MessageListener<MessageType> listener)\n    {\n      this.listener = listener;\n    }\n\n    private void start()\n    {\n      if (!watermark.compareAndSet(INIT, 0)) {\n        throw new ISE(\"Already started\");\n      }\n\n      listener.serverAdded(serverNode);\n      issueNextGetMessagesCall();\n    }\n\n    private void issueNextGetMessagesCall()\n    {\n      if (closed.get()) {\n        return;\n      }\n\n      final long theEpoch = epoch.get();\n      final long theWatermark = watermark.get();\n\n      log.debug(\n          \"Getting messages from server[%s] for client[%s] (current state: epoch[%s] watermark[%s]).\",\n          serverNode.getHostAndPortToUse(),","sourceCodeStart":91,"sourceCodeEnd":127,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/server/src/main/java/org/apache/druid/messages/client/MessageRelay.java#L91-L127","documentation":"MessageRelay's internal listener uses a watermark AtomicReference to ensure start() is idempotent: it CASes the watermark from INIT to 0. If the watermark is no longer INIT, the relay has already been started, and calling start() again throws ISE('Already started'). This protects against double registration of the server listener and duplicate message fetch loops.","triggerScenarios":"Calling the private start() twice on the same MessageRelay instance — e.g. a lifecycle supervisor restart that does not recreate the relay object, or racing start() calls from two threads where the second loses the CAS.","commonSituations":"Server lifecycle restart/reconfiguration paths that re-invoke start without constructing a new MessageRelay; concurrent initialization during coordinator leadership transitions.","solutions":["Ensure start() is invoked exactly once per MessageRelay instance; recreate the relay object before restarting","Guard the restart path so lifecycle stop() fully resets or replaces the relay before start() runs","Serialize initialization with a single owner (e.g. lifecycle start/stop management) to avoid concurrent start calls"],"exampleFix":"// before\nrelay.setListener(listener);\nrelay.start(); // called again on restart -> ISE\n// after\nif (relay != null) { relay.stop(); }\nrelay = new MessageRelay(...);\nrelay.setListener(listener);\nrelay.start();","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try { relay.start(); } catch (IllegalStateException e) { if (!e.getMessage().contains(\"Already started\")) { throw e; } log.debug(\"MessageRelay already started, ignoring\"); }","preventionTips":["Call start() only once per MessageRelay instance; recreate the object on restart","Manage the relay through Druid's Lifecycle so stop/start are paired","Avoid concurrent initialization of the same relay from multiple threads"],"tags":["java","lifecycle","idempotency","race-condition"],"backgroundTag":"invalid-state-transition","analyzedSha":"9b90983fd291f26935af934383ce360473179e4d","analyzedAt":"2026-09-07T13:32:30.957Z","contentChangedAt":"2026-09-07T13:32:30.957Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}