From 6f18ef718e047d074efc2ff53c51cd6523d6c82e Mon Sep 17 00:00:00 2001
From: "Lei, HUANG" <ratuthomm@gmail.com>
Date: Mon, 13 Jul 2026 16:40:18 +0800
Subject: [PATCH 01/10] fix(bulk-write): log stream writer timeouts

---
 .../java/io/greptime/BulkWriteManager.java    |  3 +-
 .../java/io/greptime/BulkWriteService.java    | 79 ++++++++++++++++++-
 .../io/greptime/BulkWriteServiceTest.java     | 52 ++++++++++++
 .../java/io/greptime/BulkWriteClient.java     | 33 +++++++-
 .../java/io/greptime/BulkWriteClientTest.java | 43 ++++++++++
 5 files changed, 204 insertions(+), 6 deletions(-)
 create mode 100644 ingester-bulk-protocol/src/test/java/io/greptime/BulkWriteServiceTest.java
 create mode 100644 ingester-protocol/src/test/java/io/greptime/BulkWriteClientTest.java

diff --git a/ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteManager.java b/ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteManager.java
index f837ce3..ebe47ca 100644
--- a/ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteManager.java
+++ b/ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteManager.java
@@ -142,7 +142,8 @@ public static BulkWriteManager create(
     public BulkWriteService intoBulkWriteStream(
             String table, Schema schema, long timeoutMs, int maxRequestsInFlight, CallOption... options) {
         FlightDescriptor descriptor = FlightDescriptor.path(table);
-        return new BulkWriteService(this, this.allocator, schema, descriptor, timeoutMs, maxRequestsInFlight, options);
+        return new BulkWriteService(
+                this, this.allocator, schema, descriptor, table, timeoutMs, maxRequestsInFlight, options);
     }
 
     VectorSchemaRoot createSchemaRoot(Schema schema) {
diff --git a/ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteService.java b/ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteService.java
index 610b97a..9482a72 100644
--- a/ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteService.java
+++ b/ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteService.java
@@ -23,6 +23,8 @@
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import org.apache.arrow.flight.BulkFlightClient.ClientStreamListener;
 import org.apache.arrow.flight.BulkFlightClient.PutListener;
@@ -60,6 +62,7 @@ public class BulkWriteService implements AutoCloseable {
     private final VectorSchemaRoot root;
     private final ClientStreamListener listener;
     private final AsyncPutListener metadataListener;
+    private final String tableName;
     private final long timeoutMs;
 
     /**
@@ -81,11 +84,27 @@ public BulkWriteService(
             long timeoutMs,
             int maxRequestsInFlight,
             CallOption... options) {
+        this(manager, allocator, schema, descriptor, "unknown", timeoutMs, maxRequestsInFlight, options);
+    }
+
+    /**
+     * Constructs a new BulkWriteService with the table name used for request diagnostics.
+     */
+    public BulkWriteService(
+            BulkWriteManager manager,
+            BufferAllocator allocator,
+            Schema schema,
+            FlightDescriptor descriptor,
+            String tableName,
+            long timeoutMs,
+            int maxRequestsInFlight,
+            CallOption... options) {
         this.manager = manager;
         this.allocator = allocator;
         this.root = manager.createSchemaRoot(schema);
         this.metadataListener = new AsyncPutListener();
         this.listener = manager.startPut(descriptor, this.metadataListener, maxRequestsInFlight, options);
+        this.tableName = tableName;
         this.timeoutMs = timeoutMs;
     }
 
@@ -146,12 +165,13 @@ public boolean isStreamReady() {
      */
     public PutStage putNext() {
         long id = nextId();
-        long totalRowCount = this.root.getRowCount();
+        int totalRowCount = this.root.getRowCount();
 
         LOG.debug("Starting putNext operation [id={}], total row count: {}", id, totalRowCount);
 
         // Create future with timeout and attach to listener
-        IdentifiableCompletableFuture future = new IdentifiableCompletableFuture(id, this.timeoutMs);
+        IdentifiableCompletableFuture future =
+                new IdentifiableCompletableFuture(id, this.timeoutMs, this.tableName, totalRowCount, LOG);
         this.metadataListener.attach(id, future);
 
         // Prepare metadata buffer
@@ -244,6 +264,19 @@ public CompletableFuture<Integer> future() {
         public int numInFlight() {
             return this.numInFlight;
         }
+
+        /**
+         * Logs diagnostic context when a caller times out waiting for this request.
+         *
+         * @param timeout the timeout raised to the caller
+         * @param value the caller's timeout value
+         * @param unit the caller's timeout unit
+         */
+        public void logTimeout(Throwable timeout, long value, TimeUnit unit) {
+            if (this.future instanceof IdentifiableCompletableFuture) {
+                ((IdentifiableCompletableFuture) this.future).logTimeout(timeout, unit.toMillis(value));
+            }
+        }
     }
 
     /**
@@ -252,6 +285,11 @@ public int numInFlight() {
      */
     static class IdentifiableCompletableFuture extends TimeoutCompletableFuture<Integer> {
         private final long id;
+        private final long timeoutMs;
+        private final String tableName;
+        private final int rows;
+        private final Logger logger;
+        private final AtomicBoolean timeoutLogged = new AtomicBoolean();
 
         /**
          * Creates a new IdentifiableCompletableFuture.
@@ -260,8 +298,16 @@ static class IdentifiableCompletableFuture extends TimeoutCompletableFuture<Inte
          * @param timeoutMs The timeout in milliseconds
          */
         public IdentifiableCompletableFuture(long id, long timeoutMs) {
+            this(id, timeoutMs, "unknown", 0, LOG);
+        }
+
+        IdentifiableCompletableFuture(long id, long timeoutMs, String tableName, int rows, Logger logger) {
             super(timeoutMs, TimeUnit.MILLISECONDS);
             this.id = id;
+            this.timeoutMs = timeoutMs;
+            this.tableName = tableName;
+            this.rows = rows;
+            this.logger = logger;
         }
 
         /**
@@ -272,6 +318,29 @@ public IdentifiableCompletableFuture(long id, long timeoutMs) {
         public long getId() {
             return this.id;
         }
+
+        @Override
+        public Integer get(long timeout, TimeUnit unit)
+                throws InterruptedException, ExecutionException, TimeoutException {
+            try {
+                return super.get(timeout, unit);
+            } catch (TimeoutException e) {
+                logTimeout(e, unit.toMillis(timeout));
+                throw e;
+            }
+        }
+
+        void logTimeout(Throwable timeout, long timeoutMs) {
+            if (this.timeoutLogged.compareAndSet(false, true)) {
+                this.logger.warn(
+                        "Bulk write timed out - table={}, request-id={}, rows={}, timeout={}ms",
+                        this.tableName,
+                        this.id,
+                        this.rows,
+                        timeoutMs,
+                        timeout);
+            }
+        }
     }
 
     /**
@@ -313,8 +382,10 @@ public void attach(long id, IdentifiableCompletableFuture future) {
                 this.futuresInFlight.remove(id);
 
                 if (t != null) {
-                    LOG.error("Put operation failed [id={}]: {}", id, t.getMessage(), t);
-                    if (!(t instanceof TimeoutCompletableFuture.FutureDeadlineExceededException)) {
+                    if (t instanceof TimeoutCompletableFuture.FutureDeadlineExceededException) {
+                        future.logTimeout(t, future.timeoutMs);
+                    } else {
+                        LOG.error("Put operation failed [id={}]: {}", id, t.getMessage(), t);
                         // If a put next operation fails, we complete the future with the exception
                         // and the stream will be terminated immediately to prevent further operations
                         onError(t);
diff --git a/ingester-bulk-protocol/src/test/java/io/greptime/BulkWriteServiceTest.java b/ingester-bulk-protocol/src/test/java/io/greptime/BulkWriteServiceTest.java
new file mode 100644
index 0000000..072bfa1
--- /dev/null
+++ b/ingester-bulk-protocol/src/test/java/io/greptime/BulkWriteServiceTest.java
@@ -0,0 +1,52 @@
+/*
+ * Copyright 2023 Greptime Team
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.greptime;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.slf4j.Logger;
+
+public class BulkWriteServiceTest {
+
+    @Test
+    public void testTimedGetLogsBulkWriteRequestContextOnce() throws Exception {
+        Logger logger = Mockito.mock(Logger.class);
+        BulkWriteService.IdentifiableCompletableFuture future =
+                new BulkWriteService.IdentifiableCompletableFuture(42L, 60000L, "metrics", 100, logger);
+
+        TimeoutException timeout = null;
+        try {
+            future.get(1, TimeUnit.MILLISECONDS);
+            Assert.fail("Expected timed get to fail");
+        } catch (TimeoutException e) {
+            timeout = e;
+        }
+        future.logTimeout(timeout, 60000L);
+
+        Mockito.verify(logger)
+                .warn(
+                        "Bulk write timed out - table={}, request-id={}, rows={}, timeout={}ms",
+                        "metrics",
+                        42L,
+                        100,
+                        1L,
+                        timeout);
+    }
+}
diff --git a/ingester-protocol/src/main/java/io/greptime/BulkWriteClient.java b/ingester-protocol/src/main/java/io/greptime/BulkWriteClient.java
index fc94aac..8d5c681 100644
--- a/ingester-protocol/src/main/java/io/greptime/BulkWriteClient.java
+++ b/ingester-protocol/src/main/java/io/greptime/BulkWriteClient.java
@@ -38,8 +38,10 @@
 import io.greptime.rpc.TlsOptions;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicReference;
 import org.apache.arrow.flight.FlightCallHeaders;
 import org.apache.arrow.flight.HeaderCallOption;
@@ -236,11 +238,13 @@ public CompletableFuture<Integer> writeNext() throws Exception {
                         "Stream busy with pending requests. Check `isStreamReady()` before calling `writeNext()` to avoid busy-waiting.");
             }
 
-            return this.pipelineWriteLimiter.acquireAndDo(null, () -> {
+            AtomicReference<BulkWriteService.PutStage> putStage = new AtomicReference<>();
+            CompletableFuture<Integer> result = this.pipelineWriteLimiter.acquireAndDo(null, () -> {
                 Clock clock = Clock.defaultClock();
 
                 long startPut = clock.getTick();
                 BulkWriteService.PutStage stage = this.writer.putNext();
+                putStage.set(stage);
                 InnerMetricHelper.prepareTime().update(clock.duration(startPut), TimeUnit.MILLISECONDS);
 
                 long startCall = clock.getTick();
@@ -262,6 +266,7 @@ public CompletableFuture<Integer> writeNext() throws Exception {
 
                 return future;
             });
+            return new TimeoutLoggingFuture(result, putStage.get());
         }
 
         @Override
@@ -282,6 +287,32 @@ public void close() throws Exception {
         }
     }
 
+    static class TimeoutLoggingFuture extends CompletableFuture<Integer> {
+        private final BulkWriteService.PutStage putStage;
+
+        TimeoutLoggingFuture(CompletableFuture<Integer> delegate, BulkWriteService.PutStage putStage) {
+            this.putStage = putStage;
+            delegate.whenComplete((r, t) -> {
+                if (t == null) {
+                    complete(r);
+                } else {
+                    completeExceptionally(t);
+                }
+            });
+        }
+
+        @Override
+        public Integer get(long timeout, TimeUnit unit)
+                throws InterruptedException, ExecutionException, TimeoutException {
+            try {
+                return super.get(timeout, unit);
+            } catch (TimeoutException e) {
+                this.putStage.logTimeout(e, timeout, unit);
+                throw e;
+            }
+        }
+    }
+
     /**
      * Limiter that controls the number of concurrent bulk write operations.
      * Uses a blocking policy to ensure the maximum number of in-flight requests is not exceeded.
diff --git a/ingester-protocol/src/test/java/io/greptime/BulkWriteClientTest.java b/ingester-protocol/src/test/java/io/greptime/BulkWriteClientTest.java
new file mode 100644
index 0000000..f597956
--- /dev/null
+++ b/ingester-protocol/src/test/java/io/greptime/BulkWriteClientTest.java
@@ -0,0 +1,43 @@
+/*
+ * Copyright 2023 Greptime Team
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.greptime;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+public class BulkWriteClientTest {
+
+    @Test
+    public void testTimedGetReportsCallerTimeoutToPutStage() throws Exception {
+        BulkWriteService.PutStage stage = Mockito.mock(BulkWriteService.PutStage.class);
+        CompletableFuture<Integer> future = new BulkWriteClient.TimeoutLoggingFuture(new CompletableFuture<>(), stage);
+
+        TimeoutException timeout = null;
+        try {
+            future.get(1, TimeUnit.MILLISECONDS);
+            Assert.fail("Expected timed get to fail");
+        } catch (TimeoutException e) {
+            timeout = e;
+        }
+
+        Mockito.verify(stage).logTimeout(timeout, 1, TimeUnit.MILLISECONDS);
+    }
+}

From 6106fe6781f50fa211bb6681f0dabaaed6e1216f Mon Sep 17 00:00:00 2001
From: "Lei, HUANG" <ratuthomm@gmail.com>
Date: Mon, 13 Jul 2026 16:45:08 +0800
Subject: [PATCH 02/10] release: v0.15.1-debug

---
 ingester-all/pom.xml                | 2 +-
 ingester-bulk-protocol/pom.xml      | 2 +-
 ingester-common/pom.xml             | 2 +-
 ingester-example/pom.xml            | 2 +-
 ingester-grpc/pom.xml               | 2 +-
 ingester-integration-tests/pom.xml  | 2 +-
 ingester-prometheus-metrics/pom.xml | 2 +-
 ingester-protocol/pom.xml           | 2 +-
 ingester-rpc/pom.xml                | 2 +-
 pom.xml                             | 2 +-
 10 files changed, 10 insertions(+), 10 deletions(-)

diff --git a/ingester-all/pom.xml b/ingester-all/pom.xml
index 59628af..470e968 100644
--- a/ingester-all/pom.xml
+++ b/ingester-all/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <groupId>io.greptime</groupId>
         <artifactId>greptimedb-ingester</artifactId>
-        <version>0.15.0</version>
+        <version>0.15.1-debug</version>
     </parent>
 
     <artifactId>ingester-all</artifactId>
diff --git a/ingester-bulk-protocol/pom.xml b/ingester-bulk-protocol/pom.xml
index 2746261..aa8ac90 100644
--- a/ingester-bulk-protocol/pom.xml
+++ b/ingester-bulk-protocol/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <groupId>io.greptime</groupId>
         <artifactId>greptimedb-ingester</artifactId>
-        <version>0.15.0</version>
+        <version>0.15.1-debug</version>
     </parent>
 
     <artifactId>ingester-bulk-protocol</artifactId>
diff --git a/ingester-common/pom.xml b/ingester-common/pom.xml
index 39ad459..795ad29 100644
--- a/ingester-common/pom.xml
+++ b/ingester-common/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <groupId>io.greptime</groupId>
         <artifactId>greptimedb-ingester</artifactId>
-        <version>0.15.0</version>
+        <version>0.15.1-debug</version>
     </parent>
 
     <artifactId>ingester-common</artifactId>
diff --git a/ingester-example/pom.xml b/ingester-example/pom.xml
index d574f89..da1a12a 100644
--- a/ingester-example/pom.xml
+++ b/ingester-example/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <groupId>io.greptime</groupId>
         <artifactId>greptimedb-ingester</artifactId>
-        <version>0.15.0</version>
+        <version>0.15.1-debug</version>
     </parent>
 
     <artifactId>ingester-example</artifactId>
diff --git a/ingester-grpc/pom.xml b/ingester-grpc/pom.xml
index 2e7e8a3..c5461a2 100644
--- a/ingester-grpc/pom.xml
+++ b/ingester-grpc/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <groupId>io.greptime</groupId>
         <artifactId>greptimedb-ingester</artifactId>
-        <version>0.15.0</version>
+        <version>0.15.1-debug</version>
     </parent>
 
     <artifactId>ingester-grpc</artifactId>
diff --git a/ingester-integration-tests/pom.xml b/ingester-integration-tests/pom.xml
index bcf0873..4066ed5 100644
--- a/ingester-integration-tests/pom.xml
+++ b/ingester-integration-tests/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <groupId>io.greptime</groupId>
         <artifactId>greptimedb-ingester</artifactId>
-        <version>0.15.0</version>
+        <version>0.15.1-debug</version>
     </parent>
 
     <artifactId>ingester-integration-tests</artifactId>
diff --git a/ingester-prometheus-metrics/pom.xml b/ingester-prometheus-metrics/pom.xml
index 2222178..7c0120f 100644
--- a/ingester-prometheus-metrics/pom.xml
+++ b/ingester-prometheus-metrics/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <groupId>io.greptime</groupId>
         <artifactId>greptimedb-ingester</artifactId>
-        <version>0.15.0</version>
+        <version>0.15.1-debug</version>
     </parent>
     <artifactId>ingester-prometheus-metrics</artifactId>
     <name>${project.groupId}:${project.artifactId}</name>
diff --git a/ingester-protocol/pom.xml b/ingester-protocol/pom.xml
index 05c3f38..7dbc4a6 100644
--- a/ingester-protocol/pom.xml
+++ b/ingester-protocol/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <groupId>io.greptime</groupId>
         <artifactId>greptimedb-ingester</artifactId>
-        <version>0.15.0</version>
+        <version>0.15.1-debug</version>
     </parent>
 
     <artifactId>ingester-protocol</artifactId>
diff --git a/ingester-rpc/pom.xml b/ingester-rpc/pom.xml
index d8f16cc..eab2778 100644
--- a/ingester-rpc/pom.xml
+++ b/ingester-rpc/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <groupId>io.greptime</groupId>
         <artifactId>greptimedb-ingester</artifactId>
-        <version>0.15.0</version>
+        <version>0.15.1-debug</version>
     </parent>
 
     <artifactId>ingester-rpc</artifactId>
diff --git a/pom.xml b/pom.xml
index 4714f05..01fd210 100644
--- a/pom.xml
+++ b/pom.xml
@@ -21,7 +21,7 @@
 
     <groupId>io.greptime</groupId>
     <artifactId>greptimedb-ingester</artifactId>
-    <version>0.15.0</version>
+    <version>0.15.1-debug</version>
     <packaging>pom</packaging>
 
     <name>${project.groupId}:${project.artifactId}</name>

From 6f1f6c1a52af37c8dec61a4a4777a7f6b63ca50e Mon Sep 17 00:00:00 2001
From: "Lei, HUANG" <ratuthomm@gmail.com>
Date: Mon, 13 Jul 2026 17:22:19 +0800
Subject: [PATCH 03/10] ci: enable Maven Central snapshot publishing

---
 .github/workflows/mvn_publish.yml   | 10 ++++++++++
 ingester-all/pom.xml                |  2 +-
 ingester-bulk-protocol/pom.xml      |  2 +-
 ingester-common/pom.xml             |  2 +-
 ingester-example/pom.xml            |  2 +-
 ingester-grpc/pom.xml               |  2 +-
 ingester-integration-tests/pom.xml  |  2 +-
 ingester-prometheus-metrics/pom.xml |  2 +-
 ingester-protocol/pom.xml           |  2 +-
 ingester-rpc/pom.xml                |  2 +-
 pom.xml                             |  4 ++--
 11 files changed, 21 insertions(+), 11 deletions(-)

diff --git a/.github/workflows/mvn_publish.yml b/.github/workflows/mvn_publish.yml
index c60475b..cfca14b 100644
--- a/.github/workflows/mvn_publish.yml
+++ b/.github/workflows/mvn_publish.yml
@@ -17,6 +17,7 @@ name: Maven Publish
 on:
   release:
     types: [created]
+  workflow_dispatch:
 
 jobs:
   publish:
@@ -33,6 +34,15 @@ jobs:
           gpg-private-key: ${{ secrets.MAVEN_GPG_PRIVATE_KEY }} # Value of the GPG private key to import
           gpg-passphrase: MAVEN_GPG_PASSPHRASE # env variable for GPG private key passphrase
 
+      - name: Verify snapshot version
+        if: github.event_name == 'workflow_dispatch'
+        run: |
+          version="$(mvn help:evaluate -Dexpression=project.version -q -DforceStdout)"
+          if [[ "$version" != *-SNAPSHOT ]]; then
+            echo "Manual publishing requires a -SNAPSHOT version, got: $version"
+            exit 1
+          fi
+
       - name: Build with Maven
         run: mvn clean deploy --batch-mode -DskipTests -P release -B -U -e
         env:
diff --git a/ingester-all/pom.xml b/ingester-all/pom.xml
index 470e968..75e311d 100644
--- a/ingester-all/pom.xml
+++ b/ingester-all/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <groupId>io.greptime</groupId>
         <artifactId>greptimedb-ingester</artifactId>
-        <version>0.15.1-debug</version>
+        <version>0.15.1-SNAPSHOT</version>
     </parent>
 
     <artifactId>ingester-all</artifactId>
diff --git a/ingester-bulk-protocol/pom.xml b/ingester-bulk-protocol/pom.xml
index aa8ac90..12d36e2 100644
--- a/ingester-bulk-protocol/pom.xml
+++ b/ingester-bulk-protocol/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <groupId>io.greptime</groupId>
         <artifactId>greptimedb-ingester</artifactId>
-        <version>0.15.1-debug</version>
+        <version>0.15.1-SNAPSHOT</version>
     </parent>
 
     <artifactId>ingester-bulk-protocol</artifactId>
diff --git a/ingester-common/pom.xml b/ingester-common/pom.xml
index 795ad29..057fc64 100644
--- a/ingester-common/pom.xml
+++ b/ingester-common/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <groupId>io.greptime</groupId>
         <artifactId>greptimedb-ingester</artifactId>
-        <version>0.15.1-debug</version>
+        <version>0.15.1-SNAPSHOT</version>
     </parent>
 
     <artifactId>ingester-common</artifactId>
diff --git a/ingester-example/pom.xml b/ingester-example/pom.xml
index da1a12a..442536d 100644
--- a/ingester-example/pom.xml
+++ b/ingester-example/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <groupId>io.greptime</groupId>
         <artifactId>greptimedb-ingester</artifactId>
-        <version>0.15.1-debug</version>
+        <version>0.15.1-SNAPSHOT</version>
     </parent>
 
     <artifactId>ingester-example</artifactId>
diff --git a/ingester-grpc/pom.xml b/ingester-grpc/pom.xml
index c5461a2..225c7d3 100644
--- a/ingester-grpc/pom.xml
+++ b/ingester-grpc/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <groupId>io.greptime</groupId>
         <artifactId>greptimedb-ingester</artifactId>
-        <version>0.15.1-debug</version>
+        <version>0.15.1-SNAPSHOT</version>
     </parent>
 
     <artifactId>ingester-grpc</artifactId>
diff --git a/ingester-integration-tests/pom.xml b/ingester-integration-tests/pom.xml
index 4066ed5..38a205c 100644
--- a/ingester-integration-tests/pom.xml
+++ b/ingester-integration-tests/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <groupId>io.greptime</groupId>
         <artifactId>greptimedb-ingester</artifactId>
-        <version>0.15.1-debug</version>
+        <version>0.15.1-SNAPSHOT</version>
     </parent>
 
     <artifactId>ingester-integration-tests</artifactId>
diff --git a/ingester-prometheus-metrics/pom.xml b/ingester-prometheus-metrics/pom.xml
index 7c0120f..146ec59 100644
--- a/ingester-prometheus-metrics/pom.xml
+++ b/ingester-prometheus-metrics/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <groupId>io.greptime</groupId>
         <artifactId>greptimedb-ingester</artifactId>
-        <version>0.15.1-debug</version>
+        <version>0.15.1-SNAPSHOT</version>
     </parent>
     <artifactId>ingester-prometheus-metrics</artifactId>
     <name>${project.groupId}:${project.artifactId}</name>
diff --git a/ingester-protocol/pom.xml b/ingester-protocol/pom.xml
index 7dbc4a6..1d03587 100644
--- a/ingester-protocol/pom.xml
+++ b/ingester-protocol/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <groupId>io.greptime</groupId>
         <artifactId>greptimedb-ingester</artifactId>
-        <version>0.15.1-debug</version>
+        <version>0.15.1-SNAPSHOT</version>
     </parent>
 
     <artifactId>ingester-protocol</artifactId>
diff --git a/ingester-rpc/pom.xml b/ingester-rpc/pom.xml
index eab2778..0f8bc78 100644
--- a/ingester-rpc/pom.xml
+++ b/ingester-rpc/pom.xml
@@ -21,7 +21,7 @@
     <parent>
         <groupId>io.greptime</groupId>
         <artifactId>greptimedb-ingester</artifactId>
-        <version>0.15.1-debug</version>
+        <version>0.15.1-SNAPSHOT</version>
     </parent>
 
     <artifactId>ingester-rpc</artifactId>
diff --git a/pom.xml b/pom.xml
index 01fd210..0554624 100644
--- a/pom.xml
+++ b/pom.xml
@@ -21,7 +21,7 @@
 
     <groupId>io.greptime</groupId>
     <artifactId>greptimedb-ingester</artifactId>
-    <version>0.15.1-debug</version>
+    <version>0.15.1-SNAPSHOT</version>
     <packaging>pom</packaging>
 
     <name>${project.groupId}:${project.artifactId}</name>
@@ -284,7 +284,7 @@
                     <plugin>
                         <groupId>org.sonatype.central</groupId>
                         <artifactId>central-publishing-maven-plugin</artifactId>
-                        <version>0.4.0</version>
+                        <version>0.11.0</version>
                         <extensions>true</extensions>
                         <configuration>
                             <publishingServerId>central</publishingServerId>

From 37fc57cf9e01d3695d9c8e0732461d919e9578a9 Mon Sep 17 00:00:00 2001
From: "Lei, HUANG" <ratuthomm@gmail.com>
Date: Mon, 13 Jul 2026 18:50:56 +0800
Subject: [PATCH 04/10] fix: configure Central snapshot repository

---
 pom.xml | 1 +
 1 file changed, 1 insertion(+)

diff --git a/pom.xml b/pom.xml
index 0554624..882df0d 100644
--- a/pom.xml
+++ b/pom.xml
@@ -288,6 +288,7 @@
                         <extensions>true</extensions>
                         <configuration>
                             <publishingServerId>central</publishingServerId>
+                            <centralSnapshotsUrl>https://central.sonatype.com/repository/maven-snapshots/</centralSnapshotsUrl>
                             <tokenAuth>true</tokenAuth>
                         </configuration>
                     </plugin>

From f822b4f9d07cb4122d41a37612354beb07cff62a Mon Sep 17 00:00:00 2001
From: "Lei, HUANG" <ratuthomm@gmail.com>
Date: Tue, 14 Jul 2026 20:18:46 +0800
Subject: [PATCH 05/10] fix: resolve AsyncPutListener terminal-stream/attach
 race and supplier-failure permit leak

Replace the one-shot weakly consistent terminal fanout with a
lock-protected terminal state machine in `AsyncPutListener` so
`attach()` and `onError()`/`onCompleted()` cannot race. When
`putNext()` throws synchronously, the unsent Arrow metadata buffer
is released and the write-limiter permit is returned immediately.

- `ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteService.java`
- `ingester-protocol/src/main/java/io/greptime/limit/AbstractLimiter.java`
- `ingester-bulk-protocol/src/test/java/io/greptime/BulkWriteServiceTest.java`
- `ingester-protocol/src/test/java/io/greptime/WriteLimitTest.java`

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>
---
 .../java/io/greptime/BulkWriteService.java    | 138 ++++++--
 .../io/greptime/BulkWriteServiceTest.java     | 300 ++++++++++++++++++
 .../io/greptime/limit/AbstractLimiter.java    |   7 +-
 .../test/java/io/greptime/WriteLimitTest.java |  40 ++-
 4 files changed, 450 insertions(+), 35 deletions(-)

diff --git a/ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteService.java b/ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteService.java
index 9482a72..22f44df 100644
--- a/ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteService.java
+++ b/ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteService.java
@@ -18,7 +18,10 @@
 
 import com.google.protobuf.ByteString;
 import io.greptime.common.TimeoutCompletableFuture;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutionException;
@@ -172,26 +175,64 @@ public PutStage putNext() {
         // Create future with timeout and attach to listener
         IdentifiableCompletableFuture future =
                 new IdentifiableCompletableFuture(id, this.timeoutMs, this.tableName, totalRowCount, LOG);
-        this.metadataListener.attach(id, future);
-
-        // Prepare metadata buffer
-        byte[] metadata = new Metadata.RequestMetadata(id).toJsonBytesUtf8();
+        ArrowBuf metadataBuf = null;
+        Throwable putFailure = null;
         try {
-            ArrowBuf metadataBuf = this.allocator.buffer(metadata.length);
+            if (!this.metadataListener.attach(id, future)) {
+                return new PutStage(future, this.metadataListener.numInFlight());
+            }
+
+            // Prepare metadata buffer
+            byte[] metadata = new Metadata.RequestMetadata(id).toJsonBytesUtf8();
+            metadataBuf = this.allocator.buffer(metadata.length);
             metadataBuf.writeBytes(metadata);
 
             // Send data to the server
             LOG.debug("Sending data to server [id={}]", id);
             this.listener.putNext(metadataBuf);
+            metadataBuf = null; // Ownership transfers to the Flight writer.
 
             int inFlightCount = this.metadataListener.numInFlight();
             LOG.debug("Data sent successfully [id={}], in-flight requests: {}", id, inFlightCount);
 
             return new PutStage(future, inFlightCount);
+        } catch (RuntimeException | Error e) {
+            Throwable failure = e;
+            if (!future.completeExceptionally(e)) {
+                try {
+                    future.join();
+                } catch (CompletionException completedFailure) {
+                    if (completedFailure.getCause() != null) {
+                        failure = completedFailure.getCause();
+                    }
+                }
+            }
+            putFailure = failure;
+            if (metadataBuf != null && metadataBuf.refCnt() > 0) {
+                try {
+                    metadataBuf.close();
+                } catch (RuntimeException closeError) {
+                    failure.addSuppressed(closeError);
+                }
+            }
+            if (failure instanceof RuntimeException) {
+                throw (RuntimeException) failure;
+            }
+            if (failure instanceof Error) {
+                throw (Error) failure;
+            }
+            throw new CompletionException(failure);
         } finally {
             // Clear the root to prepare for next batch
-            this.root.clear();
-            LOG.debug("Cleared root for next batch [id={}], previous row count: {}", id, totalRowCount);
+            try {
+                this.root.clear();
+                LOG.debug("Cleared root for next batch [id={}], previous row count: {}", id, totalRowCount);
+            } catch (RuntimeException | Error clearError) {
+                if (putFailure == null) {
+                    throw clearError;
+                }
+                putFailure.addSuppressed(clearError);
+            }
         }
     }
 
@@ -350,6 +391,9 @@ void logTimeout(Throwable timeout, long timeoutMs) {
     static class AsyncPutListener implements PutListener {
         private final ConcurrentMap<Long, IdentifiableCompletableFuture> futuresInFlight;
         private final CompletableFuture<Void> completed;
+        private final Object terminalLock;
+        private boolean terminal;
+        private Throwable terminalFailure;
 
         /**
          * Creates a new AsyncPutListener.
@@ -357,16 +401,7 @@ static class AsyncPutListener implements PutListener {
         AsyncPutListener() {
             this.futuresInFlight = new ConcurrentHashMap<>();
             this.completed = new CompletableFuture<>();
-            this.completed.whenComplete((r, t) -> {
-                if (t != null) {
-                    // Also complete all the futures with the same exception
-                    for (IdentifiableCompletableFuture future : this.futuresInFlight.values()) {
-                        future.completeExceptionally(t);
-                    }
-                }
-                // When completed, clear the futuresInFlight
-                this.futuresInFlight.clear();
-            });
+            this.terminalLock = new Object();
         }
 
         /**
@@ -374,12 +409,12 @@ static class AsyncPutListener implements PutListener {
          *
          * @param id The unique identifier for the request
          * @param future The future to track
+         * @return true if the future was attached, false if the stream was already terminal
          */
-        public void attach(long id, IdentifiableCompletableFuture future) {
-            this.futuresInFlight.put(id, future);
+        public boolean attach(long id, IdentifiableCompletableFuture future) {
             future.whenComplete((r, t) -> {
                 // Remove the future from the map when it's completed
-                this.futuresInFlight.remove(id);
+                this.futuresInFlight.remove(id, future);
 
                 if (t != null) {
                     if (t instanceof TimeoutCompletableFuture.FutureDeadlineExceededException) {
@@ -394,11 +429,28 @@ public void attach(long id, IdentifiableCompletableFuture future) {
                     LOG.debug("Put operation succeeded [id={}], affected rows: {}", id, r);
                 }
             });
+
+            Throwable failure;
+            synchronized (this.terminalLock) {
+                if (this.terminal) {
+                    failure = this.terminalFailure;
+                } else {
+                    this.futuresInFlight.put(id, future);
+                    failure = null;
+                }
+            }
+
+            if (failure != null) {
+                future.completeExceptionally(failure);
+                return false;
+            }
+
             future.scheduleTimeout();
 
             if (LOG.isDebugEnabled()) {
                 LOG.debug("Attached future [id={}], current in-flight count: {}", id, this.futuresInFlight.size());
             }
+            return true;
         }
 
         /**
@@ -435,14 +487,52 @@ public void onNext(PutResult val) {
 
         @Override
         public void onError(Throwable t) {
-            LOG.error("Stream error occurred: {}", t.getMessage(), t);
-            this.completed.completeExceptionally(StatusUtils.fromThrowable(t));
+            Throwable failure = StatusUtils.fromThrowable(t);
+            if (terminate(failure, false)) {
+                LOG.error("Stream error occurred: {}", t.getMessage(), t);
+            }
         }
 
         @Override
         public final void onCompleted() {
-            LOG.info("Server signaled stream completion");
-            this.completed.complete(null);
+            if (terminate(null, true)) {
+                LOG.info("Server signaled stream completion");
+            }
+        }
+
+        private boolean terminate(Throwable failure, boolean normalCompletion) {
+            List<IdentifiableCompletableFuture> pending;
+            Throwable completionFailure;
+            synchronized (this.terminalLock) {
+                if (this.terminal) {
+                    return false;
+                }
+
+                pending = new ArrayList<>(this.futuresInFlight.values());
+                this.futuresInFlight.clear();
+                if (normalCompletion && pending.isEmpty()) {
+                    completionFailure = null;
+                    this.terminalFailure = new IllegalStateException("The bulk write stream is already completed");
+                } else if (failure != null) {
+                    completionFailure = failure;
+                    this.terminalFailure = failure;
+                } else {
+                    completionFailure = new IllegalStateException(
+                            "The bulk write stream completed before all put responses were received");
+                    this.terminalFailure = completionFailure;
+                }
+                this.terminal = true;
+            }
+
+            if (completionFailure != null) {
+                this.completed.completeExceptionally(completionFailure);
+                for (IdentifiableCompletableFuture future : pending) {
+                    future.completeExceptionally(completionFailure);
+                }
+            } else {
+                this.completed.complete(null);
+            }
+            return true;
         }
 
         @Override
diff --git a/ingester-bulk-protocol/src/test/java/io/greptime/BulkWriteServiceTest.java b/ingester-bulk-protocol/src/test/java/io/greptime/BulkWriteServiceTest.java
index 072bfa1..c537bac 100644
--- a/ingester-bulk-protocol/src/test/java/io/greptime/BulkWriteServiceTest.java
+++ b/ingester-bulk-protocol/src/test/java/io/greptime/BulkWriteServiceTest.java
@@ -16,15 +16,257 @@
 
 package io.greptime;
 
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.arrow.flight.BulkFlightClient.ClientStreamListener;
+import org.apache.arrow.flight.BulkFlightClient.PutListener;
+import org.apache.arrow.flight.CallOption;
+import org.apache.arrow.flight.CallStatus;
+import org.apache.arrow.flight.FlightDescriptor;
+import org.apache.arrow.flight.FlightRuntimeException;
+import org.apache.arrow.flight.FlightStatusCode;
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.pojo.Schema;
 import org.junit.Assert;
 import org.junit.Test;
+import org.mockito.ArgumentCaptor;
 import org.mockito.Mockito;
 import org.slf4j.Logger;
 
 public class BulkWriteServiceTest {
 
+    @Test
+    public void testAttachAfterStreamErrorFailsImmediately() throws Exception {
+        BulkWriteService.AsyncPutListener listener = new BulkWriteService.AsyncPutListener();
+        listener.onError(CallStatus.UNAVAILABLE.toRuntimeException());
+
+        BulkWriteService.IdentifiableCompletableFuture future = newFuture(1L);
+        listener.attach(1L, future);
+
+        Throwable failure = getFailure(future);
+        Assert.assertTrue(failure instanceof FlightRuntimeException);
+        Assert.assertEquals(
+                FlightStatusCode.UNAVAILABLE,
+                ((FlightRuntimeException) failure).status().code());
+        Assert.assertEquals(0, listener.numInFlight());
+    }
+
+    @Test
+    public void testAttachAfterNormalCompletionFailsImmediately() throws Exception {
+        BulkWriteService.AsyncPutListener listener = new BulkWriteService.AsyncPutListener();
+        listener.onCompleted();
+
+        BulkWriteService.IdentifiableCompletableFuture future = newFuture(1L);
+        listener.attach(1L, future);
+
+        Assert.assertTrue(getFailure(future) instanceof IllegalStateException);
+        Assert.assertEquals(0, listener.numInFlight());
+    }
+
+    @Test
+    public void testNormalCompletionFailsPendingFuture() throws Exception {
+        BulkWriteService.AsyncPutListener listener = new BulkWriteService.AsyncPutListener();
+        BulkWriteService.IdentifiableCompletableFuture future = newFuture(1L);
+        listener.attach(1L, future);
+
+        listener.onCompleted();
+
+        Assert.assertTrue(getFailure(future) instanceof IllegalStateException);
+        Assert.assertEquals(0, listener.numInFlight());
+        Assert.assertTrue(listener.isCompletedExceptionally());
+    }
+
+    @Test
+    public void testConcurrentAttachAndStreamErrorAlwaysCompletesFuture() throws Exception {
+        ExecutorService executor = Executors.newFixedThreadPool(2);
+        try {
+            for (int i = 0; i < 100; i++) {
+                BulkWriteService.AsyncPutListener listener = new BulkWriteService.AsyncPutListener();
+                BulkWriteService.IdentifiableCompletableFuture future = newFuture(i + 1L);
+                CountDownLatch start = new CountDownLatch(1);
+
+                Future<?> attach = executor.submit(() -> {
+                    await(start);
+                    listener.attach(future.getId(), future);
+                });
+                Future<?> terminate = executor.submit(() -> {
+                    await(start);
+                    listener.onError(CallStatus.UNAVAILABLE.toRuntimeException());
+                });
+
+                start.countDown();
+                attach.get(1, TimeUnit.SECONDS);
+                terminate.get(1, TimeUnit.SECONDS);
+
+                Assert.assertTrue("Future was left unresolved", future.isCompletedExceptionally());
+                Assert.assertEquals(0, listener.numInFlight());
+            }
+        } finally {
+            executor.shutdownNow();
+        }
+    }
+
+    @Test
+    public void testStreamResultCompletesBeforePendingFutureCallbacksRun() throws Exception {
+        BulkWriteService.AsyncPutListener listener = new BulkWriteService.AsyncPutListener();
+        BulkWriteService.IdentifiableCompletableFuture future = newFuture(1L);
+        AtomicBoolean callbackFinished = new AtomicBoolean();
+        AtomicBoolean callbackObservedStreamError = new AtomicBoolean();
+        future.whenComplete((r, t) -> {
+            try {
+                listener.getResult();
+            } catch (FlightRuntimeException e) {
+                callbackObservedStreamError.set(true);
+            }
+            callbackFinished.set(true);
+        });
+        listener.attach(1L, future);
+
+        ExecutorService executor = Executors.newSingleThreadExecutor();
+        try {
+            Future<?> terminate = executor.submit(() -> listener.onError(CallStatus.UNAVAILABLE.toRuntimeException()));
+            terminate.get(1, TimeUnit.SECONDS);
+        } finally {
+            executor.shutdownNow();
+        }
+
+        Assert.assertTrue(callbackFinished.get());
+        Assert.assertTrue(callbackObservedStreamError.get());
+    }
+
+    @Test
+    public void testPutNextDoesNotSendAfterStreamTermination() throws Exception {
+        try (BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE)) {
+            ServiceFixture fixture = newServiceFixture(allocator);
+            try (BulkWriteService service = fixture.service) {
+                fixture.metadataListener.onError(CallStatus.UNAVAILABLE.toRuntimeException());
+
+                BulkWriteService.PutStage stage = service.putNext();
+
+                Throwable failure = getFailure(stage.future());
+                Assert.assertTrue(failure instanceof FlightRuntimeException);
+                Assert.assertEquals(
+                        FlightStatusCode.UNAVAILABLE,
+                        ((FlightRuntimeException) failure).status().code());
+                Mockito.verify(fixture.stream, Mockito.never()).putNext(Mockito.any());
+                Assert.assertEquals(0, fixture.metadataListener.numInFlight());
+            }
+        }
+    }
+
+    @Test
+    public void testPutNextCleansUpFutureWhenSendFailsSynchronously() throws Exception {
+        try (BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE)) {
+            ServiceFixture fixture = newServiceFixture(allocator);
+            RuntimeException sendFailure = new RuntimeException("send failed");
+            Mockito.doThrow(sendFailure).when(fixture.stream).putNext(Mockito.any());
+
+            try (BulkWriteService service = fixture.service) {
+                try {
+                    service.putNext();
+                    Assert.fail("Expected putNext to fail");
+                } catch (RuntimeException e) {
+                    Assert.assertSame(sendFailure, e);
+                }
+
+                Assert.assertEquals(0, fixture.metadataListener.numInFlight());
+                Assert.assertTrue(fixture.metadataListener.isCompletedExceptionally());
+            }
+        }
+    }
+
+    @Test
+    public void testRootCleanupDoesNotMaskSynchronousSendFailure() throws Exception {
+        try (BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE)) {
+            BulkWriteManager manager = Mockito.mock(BulkWriteManager.class);
+            ClientStreamListener stream = Mockito.mock(ClientStreamListener.class);
+            VectorSchemaRoot root = Mockito.mock(VectorSchemaRoot.class);
+            Schema schema = new Schema(Collections.emptyList());
+            FlightDescriptor descriptor = FlightDescriptor.path("metrics");
+            RuntimeException sendFailure = new RuntimeException("send failed");
+            RuntimeException cleanupFailure = new RuntimeException("cleanup failed");
+
+            Mockito.when(manager.createSchemaRoot(schema)).thenReturn(root);
+            Mockito.when(manager.startPut(
+                            Mockito.eq(descriptor),
+                            Mockito.any(PutListener.class),
+                            Mockito.eq(1L),
+                            Mockito.<CallOption[]>any()))
+                    .thenReturn(stream);
+            Mockito.doThrow(sendFailure).when(stream).putNext(Mockito.any());
+            Mockito.doThrow(cleanupFailure).when(root).clear();
+
+            try (BulkWriteService service = new BulkWriteService(manager, allocator, schema, descriptor, 60000L, 1)) {
+                try {
+                    service.putNext();
+                    Assert.fail("Expected putNext to fail");
+                } catch (RuntimeException e) {
+                    Assert.assertSame(sendFailure, e);
+                    Assert.assertArrayEquals(new Throwable[] {cleanupFailure}, e.getSuppressed());
+                }
+            }
+        }
+    }
+
+    @Test
+    public void testPutNextDoesNotDoubleCloseConsumedMetadataWhenWriterFails() throws Exception {
+        try (BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE)) {
+            ServiceFixture fixture = newServiceFixture(allocator);
+            RuntimeException sendFailure = new RuntimeException("send failed");
+            Mockito.doAnswer(invocation -> {
+                        ArrowBuf metadata = invocation.getArgument(0);
+                        metadata.close();
+                        Assert.assertEquals(0, metadata.refCnt());
+                        throw sendFailure;
+                    })
+                    .when(fixture.stream)
+                    .putNext(Mockito.any());
+
+            try (BulkWriteService service = fixture.service) {
+                try {
+                    service.putNext();
+                    Assert.fail("Expected putNext to fail");
+                } catch (RuntimeException e) {
+                    Assert.assertSame(sendFailure, e);
+                }
+            }
+        }
+    }
+
+    @Test
+    public void testTerminalFailureWinsOverLaterSynchronousWriterFailure() throws Exception {
+        try (BufferAllocator allocator = new RootAllocator(Long.MAX_VALUE)) {
+            ServiceFixture fixture = newServiceFixture(allocator);
+            RuntimeException writerFailure = new RuntimeException("writer failed");
+            Mockito.doAnswer(invocation -> {
+                        fixture.metadataListener.onError(CallStatus.UNAVAILABLE.toRuntimeException());
+                        throw writerFailure;
+                    })
+                    .when(fixture.stream)
+                    .putNext(Mockito.any());
+
+            try (BulkWriteService service = fixture.service) {
+                try {
+                    service.putNext();
+                    Assert.fail("Expected putNext to fail");
+                } catch (FlightRuntimeException e) {
+                    Assert.assertEquals(FlightStatusCode.UNAVAILABLE, e.status().code());
+                    Assert.assertNotSame(writerFailure, e);
+                }
+            }
+        }
+    }
+
     @Test
     public void testTimedGetLogsBulkWriteRequestContextOnce() throws Exception {
         Logger logger = Mockito.mock(Logger.class);
@@ -49,4 +291,62 @@ public void testTimedGetLogsBulkWriteRequestContextOnce() throws Exception {
                         1L,
                         timeout);
     }
+
+    private static BulkWriteService.IdentifiableCompletableFuture newFuture(long id) {
+        return new BulkWriteService.IdentifiableCompletableFuture(id, TimeUnit.MINUTES.toMillis(1));
+    }
+
+    private static Throwable getFailure(java.util.concurrent.CompletableFuture<Integer> future) throws Exception {
+        try {
+            future.get(1, TimeUnit.SECONDS);
+            Assert.fail("Expected future to fail");
+            return null;
+        } catch (ExecutionException e) {
+            return e.getCause();
+        }
+    }
+
+    private static void await(CountDownLatch latch) {
+        try {
+            latch.await();
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new RuntimeException(e);
+        }
+    }
+
+    private static ServiceFixture newServiceFixture(BufferAllocator allocator) {
+        BulkWriteManager manager = Mockito.mock(BulkWriteManager.class);
+        ClientStreamListener stream = Mockito.mock(ClientStreamListener.class);
+        Schema schema = new Schema(Collections.emptyList());
+        FlightDescriptor descriptor = FlightDescriptor.path("metrics");
+        VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator);
+        ArgumentCaptor<PutListener> metadataListener = ArgumentCaptor.forClass(PutListener.class);
+
+        Mockito.when(manager.createSchemaRoot(schema)).thenReturn(root);
+        Mockito.when(manager.startPut(
+                        Mockito.eq(descriptor),
+                        metadataListener.capture(),
+                        Mockito.eq(1L),
+                        Mockito.<CallOption[]>any()))
+                .thenReturn(stream);
+
+        BulkWriteService service = new BulkWriteService(manager, allocator, schema, descriptor, 60000L, 1);
+        return new ServiceFixture(service, stream, (BulkWriteService.AsyncPutListener) metadataListener.getValue());
+    }
+
+    private static class ServiceFixture {
+        private final BulkWriteService service;
+        private final ClientStreamListener stream;
+        private final BulkWriteService.AsyncPutListener metadataListener;
+
+        private ServiceFixture(
+                BulkWriteService service,
+                ClientStreamListener stream,
+                BulkWriteService.AsyncPutListener metadataListener) {
+            this.service = service;
+            this.stream = stream;
+            this.metadataListener = metadataListener;
+        }
+    }
 }
diff --git a/ingester-protocol/src/main/java/io/greptime/limit/AbstractLimiter.java b/ingester-protocol/src/main/java/io/greptime/limit/AbstractLimiter.java
index 1bb5c88..7ebc0f0 100644
--- a/ingester-protocol/src/main/java/io/greptime/limit/AbstractLimiter.java
+++ b/ingester-protocol/src/main/java/io/greptime/limit/AbstractLimiter.java
@@ -68,7 +68,12 @@ public CompletableFuture<Out> acquireAndDo(In in, Supplier<CompletableFuture<Out
 
         try {
             if (this.policy.acquire(this.limiter, permits)) {
-                return action.get().whenComplete((r, e) -> release(permits));
+                try {
+                    return action.get().whenComplete((r, e) -> release(permits));
+                } catch (RuntimeException | Error e) {
+                    release(permits);
+                    throw e;
+                }
             }
             return Util.completedCf(rejected(in, acquirePermits, maxPermits));
         } finally {
diff --git a/ingester-protocol/src/test/java/io/greptime/WriteLimitTest.java b/ingester-protocol/src/test/java/io/greptime/WriteLimitTest.java
index 274d7ac..459d3e4 100644
--- a/ingester-protocol/src/test/java/io/greptime/WriteLimitTest.java
+++ b/ingester-protocol/src/test/java/io/greptime/WriteLimitTest.java
@@ -25,9 +25,10 @@
 import io.greptime.models.WriteOk;
 import java.util.Collection;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -61,6 +62,24 @@ public void discardWriteLimitTest() throws ExecutionException, InterruptedExcept
         Assert.assertEquals(Result.FLOW_CONTROL, ret.getErr().getCode());
     }
 
+    @Test
+    public void supplierFailureReleasesWritePermit() throws Exception {
+        WriteLimiter limiter = new WriteClient.DefaultWriteLimiter(1, new LimitedPolicy.AbortPolicy());
+        Collection<Table> rows = TestUtil.testTable("test1", 1);
+        RuntimeException failure = new RuntimeException("write failed");
+
+        try {
+            limiter.acquireAndDo(rows, () -> {
+                throw failure;
+            });
+            Assert.fail("Expected write action to fail");
+        } catch (RuntimeException e) {
+            Assert.assertSame(failure, e);
+        }
+
+        Assert.assertTrue(limiter.acquireAndDo(rows, this::emptyOk).get().isOk());
+    }
+
     @Test
     public void blockingWriteLimitTest() throws InterruptedException {
         WriteLimiter limiter = new WriteClient.DefaultWriteLimiter(1, new LimitedPolicy.BlockingPolicy());
@@ -69,24 +88,25 @@ public void blockingWriteLimitTest() throws InterruptedException {
         // consume the permits
         limiter.acquireAndDo(rows, CompletableFuture::new);
 
-        final AtomicBoolean alwaysFalse = new AtomicBoolean();
+        CountDownLatch acquiring = new CountDownLatch(1);
+        AtomicReference<Throwable> failure = new AtomicReference<>();
         final Thread t = new Thread(() -> {
+            acquiring.countDown();
             try {
                 limiter.acquireAndDo(rows, this::emptyOk);
-                alwaysFalse.set(true);
             } catch (Throwable err) {
-                // noinspection ConstantConditions
-                Assert.assertTrue(err instanceof InterruptedException);
+                failure.set(err);
             }
         });
         t.start();
 
-        Assert.assertFalse(alwaysFalse.get());
-        Thread.sleep(1000);
-        Assert.assertFalse(alwaysFalse.get());
+        acquiring.await();
         t.interrupt();
-        Assert.assertFalse(alwaysFalse.get());
-        Assert.assertTrue(t.isInterrupted());
+        t.join(TimeUnit.SECONDS.toMillis(1));
+
+        Assert.assertFalse("Limiter thread did not stop after interruption", t.isAlive());
+        Assert.assertTrue(failure.get() instanceof LimitedException);
+        Assert.assertTrue(failure.get().getCause() instanceof InterruptedException);
     }
 
     @Test

From 7ee21b4d8e475ae6e4386f72968ebe5a1b308aa0 Mon Sep 17 00:00:00 2001
From: "Lei, HUANG" <ratuthomm@gmail.com>
Date: Wed, 15 Jul 2026 17:47:47 +0800
Subject: [PATCH 06/10] fix: avoid duplicate errors for terminal writes

---
 .../java/io/greptime/BulkWriteService.java    | 30 +++++++++----------
 .../io/greptime/BulkWriteServiceTest.java     |  3 +-
 2 files changed, 17 insertions(+), 16 deletions(-)

diff --git a/ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteService.java b/ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteService.java
index 22f44df..eadf96f 100644
--- a/ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteService.java
+++ b/ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteService.java
@@ -412,6 +412,21 @@ static class AsyncPutListener implements PutListener {
          * @return true if the future was attached, false if the stream was already terminal
          */
         public boolean attach(long id, IdentifiableCompletableFuture future) {
+            Throwable failure;
+            synchronized (this.terminalLock) {
+                if (this.terminal) {
+                    failure = this.terminalFailure;
+                } else {
+                    this.futuresInFlight.put(id, future);
+                    failure = null;
+                }
+            }
+
+            if (failure != null) {
+                future.completeExceptionally(failure);
+                return false;
+            }
+
             future.whenComplete((r, t) -> {
                 // Remove the future from the map when it's completed
                 this.futuresInFlight.remove(id, future);
@@ -430,21 +445,6 @@ public boolean attach(long id, IdentifiableCompletableFuture future) {
                 }
             });
 
-            Throwable failure;
-            synchronized (this.terminalLock) {
-                if (this.terminal) {
-                    failure = this.terminalFailure;
-                } else {
-                    this.futuresInFlight.put(id, future);
-                    failure = null;
-                }
-            }
-
-            if (failure != null) {
-                future.completeExceptionally(failure);
-                return false;
-            }
-
             future.scheduleTimeout();
 
             if (LOG.isDebugEnabled()) {
diff --git a/ingester-bulk-protocol/src/test/java/io/greptime/BulkWriteServiceTest.java b/ingester-bulk-protocol/src/test/java/io/greptime/BulkWriteServiceTest.java
index c537bac..7933763 100644
--- a/ingester-bulk-protocol/src/test/java/io/greptime/BulkWriteServiceTest.java
+++ b/ingester-bulk-protocol/src/test/java/io/greptime/BulkWriteServiceTest.java
@@ -47,7 +47,7 @@ public class BulkWriteServiceTest {
 
     @Test
     public void testAttachAfterStreamErrorFailsImmediately() throws Exception {
-        BulkWriteService.AsyncPutListener listener = new BulkWriteService.AsyncPutListener();
+        BulkWriteService.AsyncPutListener listener = Mockito.spy(new BulkWriteService.AsyncPutListener());
         listener.onError(CallStatus.UNAVAILABLE.toRuntimeException());
 
         BulkWriteService.IdentifiableCompletableFuture future = newFuture(1L);
@@ -59,6 +59,7 @@ public void testAttachAfterStreamErrorFailsImmediately() throws Exception {
                 FlightStatusCode.UNAVAILABLE,
                 ((FlightRuntimeException) failure).status().code());
         Assert.assertEquals(0, listener.numInFlight());
+        Mockito.verify(listener, Mockito.times(1)).onError(Mockito.any());
     }
 
     @Test

From f7d7158c335add79d062e8f2376127aa1edf4813 Mon Sep 17 00:00:00 2001
From: "Lei, HUANG" <ratuthomm@gmail.com>
Date: Wed, 15 Jul 2026 17:48:28 +0800
Subject: [PATCH 07/10] docs: clarify bulk metadata ownership

---
 .../src/main/java/io/greptime/BulkWriteService.java             | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteService.java b/ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteService.java
index eadf96f..a6fe2ca 100644
--- a/ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteService.java
+++ b/ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteService.java
@@ -208,6 +208,8 @@ public PutStage putNext() {
                 }
             }
             putFailure = failure;
+            // Flight closes metadata after accepting ownership; a positive refcount here means the
+            // failure occurred before that handoff completed and this method still owns the buffer.
             if (metadataBuf != null && metadataBuf.refCnt() > 0) {
                 try {
                     metadataBuf.close();

From d3ef759b01b1c719b8caf20c806a7488e621a92b Mon Sep 17 00:00:00 2001
From: "Lei, HUANG" <ratuthomm@gmail.com>
Date: Wed, 15 Jul 2026 17:49:38 +0800
Subject: [PATCH 08/10] refactor: derive bulk diagnostics from descriptor

---
 .../java/io/greptime/BulkWriteManager.java    |  3 +--
 .../java/io/greptime/BulkWriteService.java    | 24 +++++++------------
 .../io/greptime/BulkWriteServiceTest.java     |  8 +++++++
 3 files changed, 17 insertions(+), 18 deletions(-)

diff --git a/ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteManager.java b/ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteManager.java
index ebe47ca..f837ce3 100644
--- a/ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteManager.java
+++ b/ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteManager.java
@@ -142,8 +142,7 @@ public static BulkWriteManager create(
     public BulkWriteService intoBulkWriteStream(
             String table, Schema schema, long timeoutMs, int maxRequestsInFlight, CallOption... options) {
         FlightDescriptor descriptor = FlightDescriptor.path(table);
-        return new BulkWriteService(
-                this, this.allocator, schema, descriptor, table, timeoutMs, maxRequestsInFlight, options);
+        return new BulkWriteService(this, this.allocator, schema, descriptor, timeoutMs, maxRequestsInFlight, options);
     }
 
     VectorSchemaRoot createSchemaRoot(Schema schema) {
diff --git a/ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteService.java b/ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteService.java
index a6fe2ca..3ceeb30 100644
--- a/ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteService.java
+++ b/ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteService.java
@@ -87,30 +87,22 @@ public BulkWriteService(
             long timeoutMs,
             int maxRequestsInFlight,
             CallOption... options) {
-        this(manager, allocator, schema, descriptor, "unknown", timeoutMs, maxRequestsInFlight, options);
-    }
-
-    /**
-     * Constructs a new BulkWriteService with the table name used for request diagnostics.
-     */
-    public BulkWriteService(
-            BulkWriteManager manager,
-            BufferAllocator allocator,
-            Schema schema,
-            FlightDescriptor descriptor,
-            String tableName,
-            long timeoutMs,
-            int maxRequestsInFlight,
-            CallOption... options) {
         this.manager = manager;
         this.allocator = allocator;
         this.root = manager.createSchemaRoot(schema);
         this.metadataListener = new AsyncPutListener();
         this.listener = manager.startPut(descriptor, this.metadataListener, maxRequestsInFlight, options);
-        this.tableName = tableName;
+        this.tableName = diagnosticName(descriptor);
         this.timeoutMs = timeoutMs;
     }
 
+    static String diagnosticName(FlightDescriptor descriptor) {
+        if (descriptor.isCommand() || descriptor.getPath().isEmpty()) {
+            return "unknown";
+        }
+        return String.join("/", descriptor.getPath());
+    }
+
     /**
      * Starts the bulk write stream with default IPC options.
      */
diff --git a/ingester-bulk-protocol/src/test/java/io/greptime/BulkWriteServiceTest.java b/ingester-bulk-protocol/src/test/java/io/greptime/BulkWriteServiceTest.java
index 7933763..3af3738 100644
--- a/ingester-bulk-protocol/src/test/java/io/greptime/BulkWriteServiceTest.java
+++ b/ingester-bulk-protocol/src/test/java/io/greptime/BulkWriteServiceTest.java
@@ -45,6 +45,14 @@
 
 public class BulkWriteServiceTest {
 
+    @Test
+    public void testDiagnosticNameDerivedFromDescriptor() {
+        Assert.assertEquals(
+                "catalog/schema/metrics",
+                BulkWriteService.diagnosticName(FlightDescriptor.path("catalog", "schema", "metrics")));
+        Assert.assertEquals("unknown", BulkWriteService.diagnosticName(FlightDescriptor.command(new byte[] {1})));
+    }
+
     @Test
     public void testAttachAfterStreamErrorFailsImmediately() throws Exception {
         BulkWriteService.AsyncPutListener listener = Mockito.spy(new BulkWriteService.AsyncPutListener());

From e784f05f85ac787d9192df5f034934ffc8d1237b Mon Sep 17 00:00:00 2001
From: "Lei, HUANG" <ratuthomm@gmail.com>
Date: Wed, 15 Jul 2026 17:50:48 +0800
Subject: [PATCH 09/10] ci: run bulk protocol tests across JDKs

---
 .github/workflows/build.yml    | 17 ++++++++++++++++-
 ingester-bulk-protocol/pom.xml | 21 +++++++++++++++++++++
 2 files changed, 37 insertions(+), 1 deletion(-)

diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index 36e19fd..201e00e 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -63,6 +63,21 @@ jobs:
             || mvn --projects ingester-grpc test
             || mvn --projects ingester-grpc test)
 
+  test_ingester_bulk_protocol:
+    needs: check
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-java@v4
+        with:
+          java-version: '8'
+          distribution: 'zulu'
+      - name: Maven Test
+        run: mvn clean install -DskipTests=true -Dmaven.javadoc.skip=true -B -V
+          && (mvn --projects ingester-bulk-protocol test
+            || mvn --projects ingester-bulk-protocol test
+            || mvn --projects ingester-bulk-protocol test)
+
   test_ingester_protocol:
     needs: check
     runs-on: ubuntu-latest
@@ -79,7 +94,7 @@ jobs:
             || mvn --projects ingester-protocol test)
 
   integration_tests:
-    needs: [test_ingester_common, test_ingester_grpc, test_ingester_protocol]
+    needs: [test_ingester_common, test_ingester_grpc, test_ingester_bulk_protocol, test_ingester_protocol]
     runs-on: ubuntu-latest
     timeout-minutes: 10
     steps:
diff --git a/ingester-bulk-protocol/pom.xml b/ingester-bulk-protocol/pom.xml
index 12d36e2..8cc7e88 100644
--- a/ingester-bulk-protocol/pom.xml
+++ b/ingester-bulk-protocol/pom.xml
@@ -118,4 +118,25 @@
             <scope>test</scope>
         </dependency>
     </dependencies>
+
+    <profiles>
+        <profile>
+            <id>java9-plus</id>
+            <activation>
+                <jdk>[9,)</jdk>
+            </activation>
+            <build>
+                <plugins>
+                    <plugin>
+                        <groupId>org.apache.maven.plugins</groupId>
+                        <artifactId>maven-surefire-plugin</artifactId>
+                        <version>2.12.4</version>
+                        <configuration>
+                            <argLine>--add-opens=java.base/java.nio=ALL-UNNAMED</argLine>
+                        </configuration>
+                    </plugin>
+                </plugins>
+            </build>
+        </profile>
+    </profiles>
 </project>

From d711c287dbd52349b6c41f1590af7ee4fabad098 Mon Sep 17 00:00:00 2001
From: "Lei, HUANG" <ratuthomm@gmail.com>
Date: Wed, 15 Jul 2026 17:51:39 +0800
Subject: [PATCH 10/10] ci: reject snapshots for release publishing

---
 .github/workflows/mvn_publish.yml | 9 +++++++++
 1 file changed, 9 insertions(+)

diff --git a/.github/workflows/mvn_publish.yml b/.github/workflows/mvn_publish.yml
index cfca14b..a203aac 100644
--- a/.github/workflows/mvn_publish.yml
+++ b/.github/workflows/mvn_publish.yml
@@ -43,6 +43,15 @@ jobs:
             exit 1
           fi
 
+      - name: Verify release version
+        if: github.event_name == 'release'
+        run: |
+          version="$(mvn help:evaluate -Dexpression=project.version -q -DforceStdout)"
+          if [[ "$version" == *-SNAPSHOT ]]; then
+            echo "Release publishing requires a non-SNAPSHOT version, got: $version"
+            exit 1
+          fi
+
       - name: Build with Maven
         run: mvn clean deploy --batch-mode -DskipTests -P release -B -U -e
         env: