apache/druid · warning

Message not accepted, message buffer full

Error message

Message not accepted, message buffer full

What it means

RabbitStreamRecordSupplier buffers records from the RabbitMQ stream into an internal bounded queue during background fetch. If the queue cannot accept the next record within recordBufferOfferTimeout milliseconds, the record is rejected, this warning is logged, and background fetching is stopped — the record is not added and the stream offset is not advanced.

Solutions

1. Increase the record buffer capacity and/or recordBufferOfferTimeout in the supervisor/IO config
2. Ensure consumers call poll() regularly; check task slowness or GC pauses
3. Handle the stopped background fetch: the supplier must be restarted/seek to resume fetching
4. Check for a stuck offsetMap entry preventing consumption from draining the buffer

Example fix

// before
"recordBufferOfferTimeout": 1000,
"recordBufferSize": 1000 // too small for stream rate
// after
"recordBufferOfferTimeout": 10000,
"recordBufferSize": 100000

try { supplier.add(partition, record); } catch (RuntimeException e) { /* buffer full triggers stopBackgroundFetch(); call supplier.seek/resume before re-adding */ }

When it happens

Trigger: Downstream consumer (poll) is slower than the background fetcher: small queue capacity, long recordBufferOfferTimeout expiry due to a full buffer, or a stall in offset advancement so the buffer stays full across offers.

Common situations: Slow query tasks not polling fast enough; buffer size configured too small for a high-throughput stream; a task pause/GC pause letting the buffer fill; buffered records retained across seek causing full buffer on other partitions.

Related errors

• Can't find location to handle segment
• Could not fetch partitions for topic/stream
• != 0
• DynamicPartitionsSpec must be used for best-effort rollup
• Failed to add a row with timestamp

AI-assisted analysis of apache/druid@9b90983fd2 (2026-09-07). Data as JSON: /api/errors/c577ca551642af4f. Report an issue: GitHub.

Appendix: source

Thrown at extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/RabbitStreamRecordSupplier.java:343

   * as short as possible and thread safe
   */
  @Override
  public void handle(MessageHandler.Context context, Message message)
  {

    OrderedPartitionableRecord<String, Long, ByteEntity> currRecord;
    currRecord = new OrderedPartitionableRecord<>(
        this.superStream,
        context.stream(),
        context.offset(),
        ImmutableList.of(new ByteEntity(message.getBodyAsBinary())));

    try {
      if (!queue.offer(
          currRecord,
          this.recordBufferOfferTimeout,
          TimeUnit.MILLISECONDS)) {
        log.warn("Message not accepted, message buffer full");
        stopBackgroundFetch();
      } else {
        this.offsetMap.put(context.stream(), OffsetSpecification.offset(context.offset() + 1));
      }
    }
    catch (InterruptedException e) {
      // may happen if interrupted while BlockingQueue.offer() is waiting
      log.warn(
          e,
          "Interrupted while waiting to add record to buffer");
      stopBackgroundFetch();
    }

  }

  /**
   * optionalStartBackgroundFetch ensures that a background fetch is running
   * if this.queue is running low on records. We want to minimize thrashing

View on GitHub (pinned to 9b90983fd2)