{"record":{"id":"c577ca551642af4f","repo":"apache/druid","slug":"message-not-accepted-message-buffer-full","errorCode":null,"errorMessage":"Message not accepted, message buffer full","messagePattern":"Message not accepted, message buffer full","errorType":"console","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/RabbitStreamRecordSupplier.java","lineNumber":343,"sourceCode":"   * as short as possible and thread safe\n   */\n  @Override\n  public void handle(MessageHandler.Context context, Message message)\n  {\n\n    OrderedPartitionableRecord<String, Long, ByteEntity> currRecord;\n    currRecord = new OrderedPartitionableRecord<>(\n        this.superStream,\n        context.stream(),\n        context.offset(),\n        ImmutableList.of(new ByteEntity(message.getBodyAsBinary())));\n\n    try {\n      if (!queue.offer(\n          currRecord,\n          this.recordBufferOfferTimeout,\n          TimeUnit.MILLISECONDS)) {\n        log.warn(\"Message not accepted, message buffer full\");\n        stopBackgroundFetch();\n      } else {\n        this.offsetMap.put(context.stream(), OffsetSpecification.offset(context.offset() + 1));\n      }\n    }\n    catch (InterruptedException e) {\n      // may happen if interrupted while BlockingQueue.offer() is waiting\n      log.warn(\n          e,\n          \"Interrupted while waiting to add record to buffer\");\n      stopBackgroundFetch();\n    }\n\n  }\n\n  /**\n   * optionalStartBackgroundFetch ensures that a background fetch is running\n   * if this.queue is running low on records. We want to minimize thrashing","sourceCodeStart":325,"sourceCodeEnd":361,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/RabbitStreamRecordSupplier.java#L325-L361","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Increase the record buffer capacity and/or recordBufferOfferTimeout in the supervisor/IO config","Ensure consumers call poll() regularly; check task slowness or GC pauses","Handle the stopped background fetch: the supplier must be restarted/seek to resume fetching","Check for a stuck offsetMap entry preventing consumption from draining the buffer"],"exampleFix":"// before\n\"recordBufferOfferTimeout\": 1000,\n\"recordBufferSize\": 1000 // too small for stream rate\n// after\n\"recordBufferOfferTimeout\": 10000,\n\"recordBufferSize\": 100000","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try { supplier.add(partition, record); } catch (RuntimeException e) { /* buffer full triggers stopBackgroundFetch(); call supplier.seek/resume before re-adding */ }","preventionTips":["Size recordBufferSize to comfortably exceed stream throughput between polls","Increase recordBufferOfferTimeout for bursty workloads","Poll frequently and avoid long GC pauses in tasks","Restart/seek the supplier when background fetch stops"],"tags":["rabbitmq","stream","backpressure","buffer-full","indexing"],"backgroundTag":"backpressure-queue-overflow","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"}