{"record":{"id":"f55b2914270b1052","repo":"apache/druid","slug":"expected-sink-to-be-no-longer-writable-before-merg","errorCode":null,"errorMessage":"Expected sink to be no longer writable before mergeAndPush for segment[%s].","messagePattern":"Expected sink to be no longer writable before mergeAndPush for segment\\[(.+?)\\]\\.","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java","lineNumber":904,"sourceCode":"      final boolean useUniquePath\n  )\n  {\n    // Bail out if this sink is null or otherwise not what we expect.\n    //noinspection ObjectEquality\n    if (sinks.get(identifier) != sink) {\n      log.warn(\"Sink for segment[%s] no longer valid, bailing out of mergeAndPush.\", identifier);\n      return null;\n    }\n\n    // Use a descriptor file to indicate that pushing has completed.\n    final File persistDir = computePersistDir(identifier);\n    final File mergedTarget = new File(persistDir, \"merged\");\n    final File descriptorFile = computeDescriptorFile(identifier);\n\n    // Sanity checks\n    for (FireHydrant hydrant : sink) {\n      if (sink.isWritable()) {\n        throw new ISE(\"Expected sink to be no longer writable before mergeAndPush for segment[%s].\", identifier);\n      }\n\n      synchronized (hydrant) {\n        if (!hydrant.hasSwapped()) {\n          throw new ISE(\"Expected sink to be fully persisted before mergeAndPush for segment[%s].\", identifier);\n        }\n      }\n    }\n\n    try {\n      if (descriptorFile.exists()) {\n        // Already pushed.\n\n        if (useUniquePath) {\n          // Don't reuse the descriptor, because the caller asked for a unique path. Leave the old one as-is, since\n          // it might serve some unknown purpose.\n          log.debug(\n              \"Segment[%s] already pushed, but we want a unique path, so will push again with a new path.\",","sourceCodeStart":886,"sourceCodeEnd":922,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java#L886-L922","documentation":"Before merging and pushing a Sink, StreamAppenderator asserts the sink is no longer writable (finishWriting() was called) and all hydrants have been persisted. If the sink is still writable, the ISE is thrown because merging an actively-writable sink would produce inconsistent segment data.","triggerScenarios":"Calling mergeAndPush() on a sink without calling finishWriting() first, or after a failed/partial persist left the sink in writable state; concurrent writes still active during push; a crashed persist left hydrants unswapped.","commonSituations":"Custom ingestion code or a task calling push() while ingestion is still adding rows; Kafka tasks killed mid-persist then retried against stale in-memory state; bugs in handoff sequencing.","solutions":["Always call finishWriting() (e.g. via appenderator.close() / persistAll) before push()","Ensure persist of all hydrants completes before mergeAndPush; check hasSwapped() on each hydrant","Inspect for concurrent writers still using the sink and stop them before push","If state is unrecoverable, restart the task to rebuild sinks from persisted segments"],"exampleFix":"// before\nappenderator.push(identifiers, committer, null);\n// after\nfor (SegmentIdWithShardSpec id : identifiers) {\n  Sink sink = appenderator.getSink(id);\n  if (sink != null && sink.isWritable()) {\n    sink.finishWriting();\n  }\n}\nappenderator.persistAll(committerSupplier.get()).get();\nappenderator.push(identifiers, committer, null);","handlingStrategy":"validation","validationCode":"if (sink.isWritable()) {\n  sink.finishWriting();\n}\nif (!sink.getHydrants().stream().allMatch(FireHydrant::hasSwapped)) {\n  throw new IllegalStateException(\"Sink not fully persisted before push\");\n}","typeGuard":null,"tryCatchPattern":"try {\n  appenderator.push(ids, committer, null);\n} catch (ISE e) {\n  log.error(\"Sink state invalid before push: %s\", e.getMessage());\n  appenderator.persistAll(committer).get();\n  throw e; // restart task to recover\n}","preventionTips":["Always finishWriting()/close ingestion before calling push","Persist all hydrants (persistAll) and await completion before mergeAndPush","Avoid concurrent add() calls while a push is in flight","Model ingestion lifecycle as phases (ingest -> persist -> push) enforced by code structure"],"tags":["ingestion","appenderator","state-machine","invariant-violation"],"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-17T15:17:12.973Z"}