{"record":{"id":"6a2c6bed462b62ef","repo":"apache/druid","slug":"new-data-cannot-be-set-in-buffer-till-all-the-old","errorCode":null,"errorMessage":"New data cannot be set in buffer till all the old data has been read","messagePattern":"New data cannot be set in buffer till all the old data has been read","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SettableByteEntity.java","lineNumber":81,"sourceCode":"\n  @Override\n  public InputStream openRaw()\n  {\n    // Duplicate the entity buffer, because the stream will update its position.\n    final SettableByteBufferInputStream stream = new SettableByteBufferInputStream();\n    stream.setBuffer(entity.getBuffer().duplicate());\n    return stream;\n  }\n\n  public static final class SettableByteBufferInputStream extends InputStream\n  {\n    @Nullable\n    private ByteBufferInputStream delegate;\n\n    public void setBuffer(ByteBuffer newBuffer)\n    {\n      if (null != delegate && available() > 0) {\n        throw new IAE(\"New data cannot be set in buffer till all the old data has been read\");\n      }\n      this.delegate = new ByteBufferInputStream(newBuffer);\n    }\n\n    @Override\n    public int read()\n    {\n      Preconditions.checkNotNull(delegate, \"Buffer is not set\");\n      return delegate.read();\n    }\n\n    @Override\n    public int read(byte[] bytes, int off, int len)\n    {\n      Preconditions.checkNotNull(delegate, \"Buffer is not set\");\n      return delegate.read(bytes, off, len);\n    }\n","sourceCodeStart":63,"sourceCodeEnd":99,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SettableByteEntity.java#L63-L99","documentation":"SettableByteEntity wraps a ByteBufferInputStream delegate for reading data pushed into the entity. setBuffer replaces the buffer with new data; if the previous buffer still has unread bytes (available() > 0), replacing it would silently discard data, so an IAE is thrown.","triggerScenarios":"Calling setBuffer (via the entity's open path, e.g. a new chunk arriving from Kafka/Kinesis fetch) while the consumer has not fully read the previous buffer.","commonSituations":"Slow or stalled consumers in seekable-stream input entities; a producer pushing the next chunk before the row parser finished the previous one; downstream exceptions leaving the stream partially consumed.","solutions":["Ensure the consumer fully reads (or explicitly closes/resets) the current buffer before pushing new data","Check the consuming task for exceptions or back-pressure causing unread data","Upgrade Druid — later versions reworked this input-entity handling for seekable streams"],"exampleFix":"// before\nentity.setBuffer(nextChunk); // throws if old data remains\n// after\nif (entity.available() == 0) {\n  entity.setBuffer(nextChunk);\n}","handlingStrategy":"validation","validationCode":"if (entity.available() > 0) {\n  throw new IllegalStateException(\"Previous buffer not fully consumed; drain before setting new buffer\");\n}","typeGuard":null,"tryCatchPattern":"try {\n  entity.setBuffer(newBuffer);\n} catch (IllegalArgumentException e) {\n  // drain remaining bytes or reset entity, then retry\n  log.warn(e, \"Buffer not drained; resetting input entity\");\n  entity.reset();\n  entity.setBuffer(newBuffer);\n}","preventionTips":["Fully read or close each chunk before pushing the next","Handle consumer exceptions so streams are reset instead of left partially read","Upgrade to a Druid version with reworked seekable-stream input entities"],"tags":["buffer","streaming","backpressure"],"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"}