{"record":{"id":"bdcf362e299328e0","repo":"apache/druid","slug":"interrupted-flushing-elements-from-queue","errorCode":null,"errorMessage":"interrupted flushing elements from queue","messagePattern":"interrupted flushing elements from queue","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"extensions-contrib/ambari-metrics-emitter/src/main/java/org/apache/druid/emitter/ambari/metrics/AmbariMetricsEmitter.java","lineNumber":266,"sourceCode":"      catch (Exception e) {\n        log.error(e, e.getMessage());\n      }\n\n    }\n  }\n\n  @Override\n  public void flush()\n  {\n    synchronized (started) {\n      if (started.get()) {\n        Future future = exec.schedule(new ConsumerRunnable(), 0, TimeUnit.MILLISECONDS);\n        try {\n          future.get(DEFAULT_FLUSH_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);\n        }\n        catch (InterruptedException | ExecutionException | TimeoutException e) {\n          if (e instanceof InterruptedException) {\n            throw new RuntimeException(\"interrupted flushing elements from queue\", e);\n          }\n        }\n      }\n    }\n  }\n\n  @Override\n  public void close()\n  {\n    synchronized (started) {\n      flush();\n      exec.shutdown();\n      started.set(false);\n    }\n  }\n\n  protected static String sanitize(String namespace)\n  {","sourceCodeStart":248,"sourceCodeEnd":284,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/extensions-contrib/ambari-metrics-emitter/src/main/java/org/apache/druid/emitter/ambari/metrics/AmbariMetricsEmitter.java#L248-L284","documentation":"deserialize reads a header byte (numHashFunc), an int (bitset length), and then that many longs from the buffer. Any RuntimeException while reading (BufferUnderflowException from truncated data, IndexOutOfBoundsException from a negative length, etc.) is caught and rethrown as an IOException with message 'Unable to deserialize BloomKFilter' and the original exception as the cause. It indicates the bytes at the given position are not a complete, well-formed BloomKFilter serialization.","triggerScenarios":"Calling BloomKFilter.deserialize(ByteBuffer, position) where fewer bytes remain than the header declares (e.g. buffer was truncated, wrong position given, wrong offset/length stored), or the data is corrupt/garbage so the declared bitsetArrayLen is invalid.","commonSituations":"Storing the serialized filter with a wrong length/offset (e.g. slicing a byte[] incorrectly); passing bytes produced by a different format or Druid version; reading the filter from a partially flushed/failed write; deserializing at a nonzero position into a buffer that does not contain a filter there.","solutions":["Inspect the cause (getCause()) to see whether it is BufferUnderflow (truncated) or a bad length value, and fix the slicing/offset that produced the buffer.","Verify the serialized byte range: store and pass the exact length returned by BloomKFilter.getSerializedSize() for the filter.","Confirm the buffer was serialized with BloomKFilter.serialize by the same (or compatible) Druid version.","Wrap in try-catch for IOException and treat the filter as missing/unknown rather than crashing the query."],"exampleFix":"// before\nbyte[] slice = new byte[16]; // guessed length, too small\nBloomKFilter bf = BloomKFilter.deserialize(ByteBuffer.wrap(slice), 0); // BufferUnderflow -> IOException\n\n// after\nbyte[] bytes = filterBytes.array();\nBloomKFilter bf = BloomKFilter.deserialize(ByteBuffer.wrap(bytes, offset, BloomKFilter.getSerializedSize()), 0);","handlingStrategy":"validation","validationCode":"boolean looksLikeBloomFilter(ByteBuffer buf, int position) {\n  if (buf == null || position < 0 || buf.remaining() <= position) return false;\n  int bitsetLongs = buf.duplicate().order(ByteOrder.BIG_ENDIAN).getInt(position + 1);\n  return bitsetLongs >= 0\n      && (position + BloomKFilter.START_OF_SERIALIZED_LONGS + (long) bitsetLongs * Long.BYTES) <= buf.capacity();\n}","typeGuard":"boolean hasCompleteFilter(byte[] bytes, int offset) {\n  if (bytes == null || offset + 5 > bytes.length) return false;\n  int longs = ((bytes[offset+1] & 0xFF) << 24) | ((bytes[offset+2] & 0xFF) << 16)\n            | ((bytes[offset+3] & 0xFF) << 8) | (bytes[offset+4] & 0xFF);\n  return offset + BloomKFilter.START_OF_SERIALIZED_LONGS + (long) longs * Long.BYTES <= bytes.length;\n}","tryCatchPattern":"try {\n  return BloomKFilter.deserialize(buffer, position);\n} catch (IOException e) {\n  LOG.warn(e, \"Corrupt bloom filter bytes at position %d\", position);\n  return null; // degrade to 'unknown membership'\n}","preventionTips":["Always serialize and store the exact length from BloomKFilter.getSerializedSize().","Verify the cause of the IOException (BufferUnderflowException => truncated data) when debugging.","Never hand-slice serialized filter bytes; keep offset+length pairs together."],"tags":["bloom-filter","deserialization","corrupt-data","druid"],"backgroundTag":"invalid-argument-format","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"}