chore: minor change

Author: fengjiachun

ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteService.java

@@ -208,10 +208,10 @@ public void close() throws Exception {
     }
 
     private long nextId() {
-        long id = this.idGenerator.incrementAndGet();
-        if (id == 0) { // Skip ID 0 as it's reserved for special cases
+        long id;
+        do {
             id = this.idGenerator.incrementAndGet();
-        }
+        } while (id == 0); // Skip ID 0 as it's reserved for special cases
         return id;
     }
 

ingester-example/src/main/java/io/greptime/bench/BulkWriteBenchmark.java

@@ -47,7 +47,7 @@ public static void main(String[] args) throws Exception {
         String endpoint = SystemPropertyUtil.get("db_endpoint", "127.0.0.1:4001");
         String dbName = SystemPropertyUtil.get("db_name", "public");
         boolean zstdCompression = SystemPropertyUtil.getBool("zstd_compression", false);
-        int batchSize = SystemPropertyUtil.getInt("batch_size_per_request", 100 * 1024);
+        int batchSize = SystemPropertyUtil.getInt("batch_size_per_request", 5 * 1024);
         LOG.info("Connect to db: {}, endpoint: {}", dbName, endpoint);
         LOG.info("Using zstd compression: {}", zstdCompression);
         LOG.info("Batch size: {}", batchSize);
@@ -62,7 +62,7 @@ public static void main(String[] args) throws Exception {
                 .allocatorInitReservation(0)
                 .allocatorMaxAllocation(4 * 1024 * 1024 * 1024L)
                 .timeoutMsPerMessage(60000)
-                .maxRequestsInFlight(8)
+                .maxRequestsInFlight(32)
                 .build();
         Compression compression = zstdCompression ? Compression.Zstd : Compression.None;
         Context ctx = Context.newDefault().withCompression(compression);

ingester-example/src/main/java/io/greptime/bench/RandomTableDataProvider.java

@@ -52,7 +52,7 @@ public class RandomTableDataProvider implements TableDataProvider {
                 .build();
         // Total number of rows to generate, configurable via system property
         // Default is 1 billion rows if not specified
-        rowCount = SystemPropertyUtil.getLong("table_data_provider.row_count", 1_000_000_000L);
+        rowCount = SystemPropertyUtil.getLong("table_data_provider.row_count", 1_000_000L);
     }
 
     @Override

ingester-example/src/main/java/io/greptime/bench/StreamingWriteBenchmark.java

@@ -48,7 +48,7 @@ public static void main(String[] args) throws Exception {
         String endpoint = SystemPropertyUtil.get("db_endpoint", "127.0.0.1:4001");
         String dbName = SystemPropertyUtil.get("db_name", "public");
         boolean zstdCompression = SystemPropertyUtil.getBool("zstd_compression", false);
-        int batchSize = SystemPropertyUtil.getInt("batch_size_per_request", 100 * 1024);
+        int batchSize = SystemPropertyUtil.getInt("batch_size_per_request", 5 * 1024);
         int maxPointsPerSecond = SystemPropertyUtil.getInt("max_points_per_second", Integer.MAX_VALUE);
         LOG.info("Connect to db: {}, endpoint: {}", dbName, endpoint);
         LOG.info("Using zstd compression: {}", zstdCompression);

ingester-protocol/src/main/java/io/greptime/BulkWriteClient.java

@@ -218,7 +218,7 @@ public CompletableFuture<Integer> writeNext() throws Exception {
                     InnerMetricHelper.putTime().update(clock.duration(startCall), TimeUnit.MILLISECONDS);
                 });
 
-                LOG.debug("Write request sent successfully, in-flight requests: {}", stage.numInFlight());
+                LOG.info("Write request sent successfully, in-flight requests: {}", stage.numInFlight());
 
                 return future;
             });