{"record":{"id":"35203b2158536edb","repo":"eclipse-vertx/vert.x","slug":"this-operation-must-be-called-from-a-vert-x-thread","errorCode":null,"errorMessage":"This operation must be called from a Vert.x thread","messagePattern":"This operation must be called from a Vert\\.x thread","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"vertx-core/src/main/java/io/vertx/core/streams/impl/InboundBuffer.java","lineNumber":103,"sourceCode":"  }\n\n  public InboundBuffer(Context context, long highWaterMark) {\n    if (context == null) {\n      throw new NullPointerException(\"context must not be null\");\n    }\n    if (highWaterMark < 0) {\n      throw new IllegalArgumentException(\"highWaterMark \" + highWaterMark + \" >= 0\");\n    }\n    this.context = (ContextInternal) context;\n    this.highWaterMark = highWaterMark;\n    this.demand = Long.MAX_VALUE;\n    // empty ArrayDeque's constructor ArrayDeque allocates 16 elements; let's delay the allocation to be of the proper size\n    this.pending = null;\n  }\n\n  private void checkThread() {\n    if (!context.inThread()) {\n      throw new IllegalStateException(\"This operation must be called from a Vert.x thread\");\n    }\n  }\n\n  /**\n   * Write an {@code element} to the buffer. The element will be delivered synchronously to the handler when\n   * it is possible, otherwise it will be queued for later delivery.\n   *\n   * @param element the element to add\n   * @return {@code false} when the producer should stop writing\n   */\n  public boolean write(E element) {\n    checkThread();\n    Handler<E> handler;\n    synchronized (this) {\n      if (demand == 0L || emitting) {\n        if (pending == null) {\n          pending = new ArrayDeque<>(1);\n        }","sourceCodeStart":85,"sourceCodeEnd":121,"githubUrl":"https://github.com/eclipse-vertx/vert.x/blob/fb308bd8c3f12c79f4ae89bef67fadf6c80d036e/vertx-core/src/main/java/io/vertx/core/streams/impl/InboundBuffer.java#L85-L121","documentation":"InboundBuffer methods such as write() must run on a Vert.x thread owned by the buffer's context; checkThread() enforces this by testing context.inThread(). The buffer mutates its pending queue and demand non-thread-safely, so off-thread calls would corrupt state, and Vert.x throws IllegalStateException instead.","triggerScenarios":"Calling inboundBuffer.write(element) (or fetch/pause/resume/draining paths that delegate to write) from a plain application thread, a third-party callback thread (e.g. a JDBC or MQTT client thread), or a different Vert.x context's event-loop/worker thread than the one the buffer was created with.","commonSituations":"Writing into a ReadStream's buffer directly from a custom client's network thread; delivering results from an executor instead of ctx.runOnContext; accessing the stream after its context/worker shut down.","solutions":["Wrap the write in the buffer's context: context.runOnContext(v -> inboundBuffer.write(element))","Dispatch from the third-party callback via vertx.runOnContext or a handler invoked on a Vert.x thread","If in a Worker/startThread scope, ensure the code executes on a context thread (vertx.executeBlocking with the right context)"],"exampleFix":"// before\nexternalClient.onMessage(msg -> inboundBuffer.write(msg));\n// after\nexternalClient.onMessage(msg ->\n  context.runOnContext(v -> inboundBuffer.write(msg)));","handlingStrategy":"try-catch","validationCode":"if (ctx != null && !ctx.isRunningOnContext()) {\n  ctx.runOnContext(v -> inboundBuffer.write(element));\n} else {\n  inboundBuffer.write(element);\n}","typeGuard":null,"tryCatchPattern":"try {\n  inboundBuffer.write(element);\n} catch (IllegalStateException e) {\n  if (e.getMessage().contains(\"Vert.x thread\")) {\n    context.runOnContext(v -> inboundBuffer.write(element));\n  } else throw e;\n}","preventionTips":["Always bridge third-party callback threads with context.runOnContext","Keep all ReadStream state mutation on the owning context","Enable Vert.x assertion mode (-Dvertx.debug=true style checks) in tests to catch off-thread access early"],"tags":["threading","streams","vertx","concurrency"],"backgroundTag":"wrong-thread","analyzedSha":"fb308bd8c3f12c79f4ae89bef67fadf6c80d036e","analyzedAt":"2026-09-06T11:37:12.241Z","contentChangedAt":"2026-09-06T11:37:12.241Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}