From cefd3a967535c69c17a875d8c020bf66da9b65ce Mon Sep 17 00:00:00 2001 From: Neenu1995 Date: Tue, 14 Jul 2026 16:53:43 -0400 Subject: [PATCH 1/5] feat(bigquery-jdbc): implement non-blocking telemetry batcher and dispatcher --- .../jdbc/telemetry/v1/TelemetryBatcher.java | 304 ++++++++++++++++++ .../telemetry/v1/TelemetryConfiguration.java | 2 +- .../telemetry/v1/TelemetryBatcherTest.java | 243 ++++++++++++++ .../v1/TelemetryConfigurationTest.java | 2 +- 4 files changed, 549 insertions(+), 2 deletions(-) create mode 100644 java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryBatcher.java create mode 100644 java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryBatcherTest.java diff --git a/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryBatcher.java b/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryBatcher.java new file mode 100644 index 000000000000..8a1f3b05f443 --- /dev/null +++ b/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryBatcher.java @@ -0,0 +1,304 @@ +/* + * Copyright 2026 Google LLC + * + * 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 + * + * https://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 com.google.cloud.bigquery.jdbc.telemetry.v1; + +import com.google.cloud.bigquery.jdbc.BigQueryJdbcCustomLogger; +import com.google.protobuf.Timestamp; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import java.util.logging.Level; +import java.util.logging.Logger; + +/** + * High-performance, non-blocking telemetry event batcher and periodic dispatcher. Buffers + * connection, statement, error, and feature usage telemetry events in memory and flushes them to + * Clearcut via {@link ClearcutTransport}. + */ +final class TelemetryBatcher implements AutoCloseable { + private static final Logger logger = + new BigQueryJdbcCustomLogger(TelemetryBatcher.class.getName()); + private static final int MAX_QUEUE_SIZE = 10_000; + private static final int MAX_PAYLOAD_BYTES = 512 * 1024; // 512 KB payload size limit + + private final TelemetryConfiguration config; + private final ClearcutTransport transport; + private final DriverEnvironment driverEnvironment; + private final ScheduledExecutorService executorService; + private final boolean ownsExecutor; + + private final ConcurrentLinkedQueue connectionAttemptQueue = + new ConcurrentLinkedQueue<>(); + private final ConcurrentLinkedQueue statementExecutionQueue = + new ConcurrentLinkedQueue<>(); + private final ConcurrentLinkedQueue errorMetricQueue = new ConcurrentLinkedQueue<>(); + private final ConcurrentLinkedQueue featureUsageQueue = + new ConcurrentLinkedQueue<>(); + + private final AtomicBoolean isClosed = new AtomicBoolean(false); + private final AtomicLong currentScheduleDelayMs = new AtomicLong(-1); + private ScheduledFuture scheduledTask; + + TelemetryBatcher(TelemetryConfiguration config, ClearcutTransport transport) { + this( + config, + transport, + Executors.newSingleThreadScheduledExecutor( + r -> { + Thread t = new Thread(r, "jdbc-telemetry-batcher"); + t.setDaemon(true); + return t; + }), + true); + } + + // Package-private constructor for testing overrides + TelemetryBatcher( + TelemetryConfiguration config, + ClearcutTransport transport, + ScheduledExecutorService executorService, + boolean ownsExecutor) { + this.config = config; + this.transport = transport; + this.driverEnvironment = config != null ? config.getDriverEnvironment() : null; + this.executorService = executorService; + this.ownsExecutor = ownsExecutor; + + if (this.config != null && this.config.isEnabled()) { + reschedule(this.config.getUploadIntervalMs()); + } + } + + void offerConnectionAttempt(ConnectionAttempt attempt) { + offerMetric(connectionAttemptQueue, attempt, "Connection attempt"); + } + + void offerStatementExecution(StatementExecution execution) { + offerMetric(statementExecutionQueue, execution, "Statement execution"); + } + + void offerErrorMetric(ErrorMetric errorMetric) { + offerMetric(errorMetricQueue, errorMetric, "Error metric"); + } + + void offerFeatureUsage(FeatureUsage featureUsage) { + offerMetric(featureUsageQueue, featureUsage, "Feature usage"); + } + + private void offerMetric(ConcurrentLinkedQueue queue, T item, String metricName) { + if (isClosed.get() || !isConfigured() || item == null) { + return; + } + if (queue.size() < MAX_QUEUE_SIZE) { + queue.offer(item); + } else { + logger.log(Level.WARNING, metricName + " telemetry queue full. Dropping metric."); + } + } + + synchronized TransportResult flush() { + if (!isConfigured() || isEmpty()) { + return TransportResult.disabled(); + } + + int maxTotalBatchSize = + config != null + ? config.getBatchSizeThreshold() + : TelemetryConfiguration.DEFAULT_BATCH_SIZE_THRESHOLD; + + List connectionAttempts = new ArrayList<>(); + List statementExecutions = new ArrayList<>(); + List errorMetrics = new ArrayList<>(); + List featureUsages = new ArrayList<>(); + + int remainingQuota = maxTotalBatchSize; + remainingQuota -= drainQueue(connectionAttemptQueue, connectionAttempts, remainingQuota); + remainingQuota -= drainQueue(statementExecutionQueue, statementExecutions, remainingQuota); + remainingQuota -= drainQueue(errorMetricQueue, errorMetrics, remainingQuota); + remainingQuota -= drainQueue(featureUsageQueue, featureUsages, remainingQuota); + + if (connectionAttempts.isEmpty() + && statementExecutions.isEmpty() + && errorMetrics.isEmpty() + && featureUsages.isEmpty()) { + return TransportResult.disabled(); + } + + Instant now = Instant.now(); + Timestamp timestamp = + Timestamp.newBuilder().setSeconds(now.getEpochSecond()).setNanos(now.getNano()).build(); + + TelemetryPayload.Builder payloadBuilder = TelemetryPayload.newBuilder().setEventTime(timestamp); + if (driverEnvironment != null) { + payloadBuilder.setDriverEnvironment(driverEnvironment); + } + payloadBuilder.addAllConnectionAttempts(connectionAttempts); + payloadBuilder.addAllStatementExecutions(statementExecutions); + payloadBuilder.addAllErrors(errorMetrics); + payloadBuilder.addAllFeatureUsages(featureUsages); + + TelemetryPayload payload = payloadBuilder.build(); + + // Enforce MAX_PAYLOAD_BYTES (512 KB) by trimming excess items if needed + while (payload.getSerializedSize() > MAX_PAYLOAD_BYTES) { + if (!featureUsages.isEmpty()) { + featureUsageQueue.offer(featureUsages.remove(featureUsages.size() - 1)); + payloadBuilder.clearFeatureUsages().addAllFeatureUsages(featureUsages); + } else if (!errorMetrics.isEmpty()) { + errorMetricQueue.offer(errorMetrics.remove(errorMetrics.size() - 1)); + payloadBuilder.clearErrors().addAllErrors(errorMetrics); + } else if (!statementExecutions.isEmpty()) { + statementExecutionQueue.offer(statementExecutions.remove(statementExecutions.size() - 1)); + payloadBuilder.clearStatementExecutions().addAllStatementExecutions(statementExecutions); + } else if (!connectionAttempts.isEmpty()) { + connectionAttemptQueue.offer(connectionAttempts.remove(connectionAttempts.size() - 1)); + payloadBuilder.clearConnectionAttempts().addAllConnectionAttempts(connectionAttempts); + } else { + break; + } + payload = payloadBuilder.build(); + } + + TransportResult result; + try { + result = transport.send(payload); + } catch (Throwable t) { + logger.log(Level.WARNING, "Unexpected exception during telemetry flush", t); + result = new TransportResult(false, -1); + } + + if (!result.isSuccess()) { + requeueItems(connectionAttemptQueue, connectionAttempts); + requeueItems(statementExecutionQueue, statementExecutions); + requeueItems(errorMetricQueue, errorMetrics); + requeueItems(featureUsageQueue, featureUsages); + } + + long uploadIntervalMs = config != null ? config.getUploadIntervalMs() : 300_000L; + long newDelayMs = + result.getNextRequestWaitMillis() > 0 + ? Math.max(uploadIntervalMs, result.getNextRequestWaitMillis()) + : uploadIntervalMs; + reschedule(newDelayMs); + + return result; + } + + @Override + public void close() { + if (!isClosed.compareAndSet(false, true)) { + return; + } + + synchronized (this) { + if (scheduledTask != null) { + scheduledTask.cancel(false); + } + } + + try { + flush(); + } catch (Throwable t) { + logger.log(Level.WARNING, "Error during final telemetry batcher flush", t); + } + + if (ownsExecutor && executorService != null) { + executorService.shutdown(); + try { + if (!executorService.awaitTermination(3, TimeUnit.SECONDS)) { + executorService.shutdownNow(); + } + } catch (InterruptedException e) { + executorService.shutdownNow(); + Thread.currentThread().interrupt(); + } + } + } + + boolean isEmpty() { + return connectionAttemptQueue.isEmpty() + && statementExecutionQueue.isEmpty() + && errorMetricQueue.isEmpty() + && featureUsageQueue.isEmpty(); + } + + int getPendingEventCount() { + return connectionAttemptQueue.size() + + statementExecutionQueue.size() + + errorMetricQueue.size() + + featureUsageQueue.size(); + } + + private boolean isConfigured() { + return config != null && config.isEnabled() && transport != null; + } + + private synchronized void reschedule(long delayMs) { + if (isClosed.get() || executorService == null || executorService.isShutdown()) { + return; + } + long effectiveDelay = Math.max(1L, delayMs); + if (currentScheduleDelayMs.get() == effectiveDelay + && scheduledTask != null + && !scheduledTask.isDone()) { + return; + } + + if (scheduledTask != null) { + scheduledTask.cancel(false); + } + + currentScheduleDelayMs.set(effectiveDelay); + scheduledTask = + executorService.scheduleWithFixedDelay( + () -> { + try { + flush(); + } catch (Throwable t) { + logger.log(Level.WARNING, "Periodic telemetry flush encountered an exception", t); + } + }, + effectiveDelay, + effectiveDelay, + TimeUnit.MILLISECONDS); + } + + private int drainQueue(ConcurrentLinkedQueue queue, List targetList, int maxItems) { + int count = 0; + T item; + while (count < maxItems && (item = queue.poll()) != null) { + targetList.add(item); + count++; + } + return count; + } + + private void requeueItems(ConcurrentLinkedQueue queue, List items) { + for (T item : items) { + if (queue.size() < MAX_QUEUE_SIZE) { + queue.offer(item); + } + } + } +} diff --git a/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryConfiguration.java b/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryConfiguration.java index 715b51c4003b..b1e54b66085b 100644 --- a/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryConfiguration.java +++ b/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryConfiguration.java @@ -25,7 +25,7 @@ final class TelemetryConfiguration { static final int DEFAULT_LOG_SOURCE = -1; static final String DEFAULT_ENDPOINT_URL = "https://play.googleapis.com/log"; static final long DEFAULT_UPLOAD_INTERVAL_MS = 300_000L; - static final int DEFAULT_BATCH_SIZE_THRESHOLD = 100; + static final int DEFAULT_BATCH_SIZE_THRESHOLD = 5000; private final boolean enabled; private final int logSource; diff --git a/java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryBatcherTest.java b/java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryBatcherTest.java new file mode 100644 index 000000000000..d3a5d97b8bf5 --- /dev/null +++ b/java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryBatcherTest.java @@ -0,0 +1,243 @@ +/* + * Copyright 2026 Google LLC + * + * 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 + * + * https://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 com.google.cloud.bigquery.jdbc.telemetry.v1; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.google.api.client.http.LowLevelHttpRequest; +import com.google.api.client.http.LowLevelHttpResponse; +import com.google.api.client.testing.http.MockHttpTransport; +import com.google.api.client.testing.http.MockLowLevelHttpRequest; +import com.google.api.client.testing.http.MockLowLevelHttpResponse; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +public class TelemetryBatcherTest { + + private ScheduledExecutorService executorService; + private DriverEnvironment driverEnvironment; + + @BeforeEach + public void setUp() { + executorService = Executors.newSingleThreadScheduledExecutor(); + driverEnvironment = + DriverEnvironment.newBuilder() + .setDriverName("SimulatedDriver") + .setDriverVersion("1.0.0") + .build(); + } + + @AfterEach + public void tearDown() { + if (executorService != null && !executorService.isShutdown()) { + executorService.shutdownNow(); + } + } + + @Test + public void testOfferAndFlushSuccess() { + AtomicInteger requestCount = new AtomicInteger(0); + MockHttpTransport mockTransport = + new MockHttpTransport() { + @Override + public LowLevelHttpRequest buildRequest(String method, String url) { + requestCount.incrementAndGet(); + return new MockLowLevelHttpRequest(url) { + @Override + public LowLevelHttpResponse execute() { + MockLowLevelHttpResponse response = new MockLowLevelHttpResponse(); + response.setStatusCode(200); + return response; + } + }; + } + }; + + TelemetryConfiguration config = + TelemetryConfiguration.newBuilder() + .setEnabled(true) + .setDriverEnvironment(driverEnvironment) + .build(); + ClearcutTransport transport = new ClearcutTransport(mockTransport, config); + + try (TelemetryBatcher batcher = + new TelemetryBatcher(config, transport, executorService, false)) { + batcher.offerConnectionAttempt( + ConnectionAttempt.newBuilder().setStatus(Status.STATUS_SUCCESS).build()); + batcher.offerStatementExecution( + StatementExecution.newBuilder().setStatus(Status.STATUS_SUCCESS).build()); + batcher.offerErrorMetric( + ErrorMetric.newBuilder().setErrorCode("ERR_001").setCount(1).build()); + batcher.offerFeatureUsage( + FeatureUsage.newBuilder().setDriverFeature(DriverFeature.DRIVER_FEATURE_CUSTOM).build()); + + assertEquals(4, batcher.getPendingEventCount()); + assertFalse(batcher.isEmpty()); + + TransportResult result = batcher.flush(); + assertTrue(result.isSuccess()); + assertEquals(1, requestCount.get()); + assertEquals(0, batcher.getPendingEventCount()); + assertTrue(batcher.isEmpty()); + } + } + + @Test + public void testFlushFailureRequeuesEvents() { + AtomicInteger requestCount = new AtomicInteger(0); + MockHttpTransport mockTransport = + new MockHttpTransport() { + @Override + public LowLevelHttpRequest buildRequest(String method, String url) { + requestCount.incrementAndGet(); + return new MockLowLevelHttpRequest(url) { + @Override + public LowLevelHttpResponse execute() { + MockLowLevelHttpResponse response = new MockLowLevelHttpResponse(); + response.setStatusCode(500); + return response; + } + }; + } + }; + + TelemetryConfiguration config = + TelemetryConfiguration.newBuilder() + .setEnabled(true) + .setDriverEnvironment(driverEnvironment) + .build(); + ClearcutTransport transport = new ClearcutTransport(mockTransport, config); + + try (TelemetryBatcher batcher = + new TelemetryBatcher(config, transport, executorService, false)) { + batcher.offerConnectionAttempt( + ConnectionAttempt.newBuilder().setStatus(Status.STATUS_ERROR).build()); + + assertEquals(1, batcher.getPendingEventCount()); + + TransportResult result = batcher.flush(); + assertFalse(result.isSuccess()); + assertEquals(1, requestCount.get()); + + // Events should be retained for next retry attempt + assertEquals(1, batcher.getPendingEventCount()); + assertFalse(batcher.isEmpty()); + } + } + + @Test + public void testBatchSizeAndPayloadLimitTrimming() { + AtomicInteger requestCount = new AtomicInteger(0); + MockHttpTransport mockTransport = + new MockHttpTransport() { + @Override + public LowLevelHttpRequest buildRequest(String method, String url) { + requestCount.incrementAndGet(); + return new MockLowLevelHttpRequest(url) { + @Override + public LowLevelHttpResponse execute() { + MockLowLevelHttpResponse response = new MockLowLevelHttpResponse(); + response.setStatusCode(200); + return response; + } + }; + } + }; + + TelemetryConfiguration config = + TelemetryConfiguration.newBuilder() + .setEnabled(true) + .setBatchSizeThreshold(5) + .setDriverEnvironment(driverEnvironment) + .build(); + ClearcutTransport transport = new ClearcutTransport(mockTransport, config); + + try (TelemetryBatcher batcher = + new TelemetryBatcher(config, transport, executorService, false)) { + for (int i = 0; i < 12; i++) { + batcher.offerConnectionAttempt( + ConnectionAttempt.newBuilder().setStatus(Status.STATUS_SUCCESS).build()); + } + + assertEquals(12, batcher.getPendingEventCount()); + + // First flush drains up to batchSizeThreshold (5 items) + TransportResult result1 = batcher.flush(); + assertTrue(result1.isSuccess()); + assertEquals(7, batcher.getPendingEventCount()); + + // Second flush drains another batchSizeThreshold (5 items) + TransportResult result2 = batcher.flush(); + assertTrue(result2.isSuccess()); + assertEquals(2, batcher.getPendingEventCount()); + + // Third flush drains remaining 2 items + TransportResult result3 = batcher.flush(); + assertTrue(result3.isSuccess()); + assertEquals(0, batcher.getPendingEventCount()); + } + } + + @Test + public void testCloseFlushesAndShutsDown() { + AtomicInteger requestCount = new AtomicInteger(0); + MockHttpTransport mockTransport = + new MockHttpTransport() { + @Override + public LowLevelHttpRequest buildRequest(String method, String url) { + requestCount.incrementAndGet(); + return new MockLowLevelHttpRequest(url) { + @Override + public LowLevelHttpResponse execute() { + MockLowLevelHttpResponse response = new MockLowLevelHttpResponse(); + response.setStatusCode(200); + return response; + } + }; + } + }; + + TelemetryConfiguration config = + TelemetryConfiguration.newBuilder() + .setEnabled(true) + .setDriverEnvironment(driverEnvironment) + .build(); + ClearcutTransport transport = new ClearcutTransport(mockTransport, config); + + TelemetryBatcher batcher = new TelemetryBatcher(config, transport); + batcher.offerConnectionAttempt( + ConnectionAttempt.newBuilder().setStatus(Status.STATUS_SUCCESS).build()); + + assertEquals(1, batcher.getPendingEventCount()); + + batcher.close(); + + assertEquals(1, requestCount.get()); + assertEquals(0, batcher.getPendingEventCount()); + + // Should not accept new events after close + batcher.offerConnectionAttempt( + ConnectionAttempt.newBuilder().setStatus(Status.STATUS_SUCCESS).build()); + assertEquals(0, batcher.getPendingEventCount()); + } +} diff --git a/java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryConfigurationTest.java b/java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryConfigurationTest.java index 1f55bd0ad87d..1d9ca1102841 100644 --- a/java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryConfigurationTest.java +++ b/java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryConfigurationTest.java @@ -34,7 +34,7 @@ public void testDefaultValues() { assertEquals(-1, config.getLogSource()); assertEquals("https://play.googleapis.com/log", config.getEndpointUrl()); assertEquals(300_000L, config.getUploadIntervalMs()); - assertEquals(100, config.getBatchSizeThreshold()); + assertEquals(5000, config.getBatchSizeThreshold()); assertNull(config.getDriverEnvironment()); } From ffe6faf87f486bfc54a39a57b49b9dcd30537007 Mon Sep 17 00:00:00 2001 From: Neenu1995 Date: Tue, 14 Jul 2026 17:18:15 -0400 Subject: [PATCH 2/5] single pass bulk payload trimming --- .../jdbc/telemetry/v1/TelemetryBatcher.java | 254 +++++++++++------- 1 file changed, 154 insertions(+), 100 deletions(-) diff --git a/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryBatcher.java b/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryBatcher.java index 8a1f3b05f443..cd65b9e899a7 100644 --- a/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryBatcher.java +++ b/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryBatcher.java @@ -28,6 +28,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.locks.ReentrantLock; import java.util.logging.Level; import java.util.logging.Logger; @@ -47,6 +48,7 @@ final class TelemetryBatcher implements AutoCloseable { private final DriverEnvironment driverEnvironment; private final ScheduledExecutorService executorService; private final boolean ownsExecutor; + private final ReentrantLock flushLock = new ReentrantLock(); private final ConcurrentLinkedQueue connectionAttemptQueue = new ConcurrentLinkedQueue<>(); @@ -117,92 +119,109 @@ private void offerMetric(ConcurrentLinkedQueue queue, T item, String metr } } - synchronized TransportResult flush() { - if (!isConfigured() || isEmpty()) { - return TransportResult.disabled(); - } + TransportResult flush() { + flushLock.lock(); + try { + if (!isConfigured() || isEmpty()) { + return TransportResult.disabled(); + } - int maxTotalBatchSize = - config != null - ? config.getBatchSizeThreshold() - : TelemetryConfiguration.DEFAULT_BATCH_SIZE_THRESHOLD; - - List connectionAttempts = new ArrayList<>(); - List statementExecutions = new ArrayList<>(); - List errorMetrics = new ArrayList<>(); - List featureUsages = new ArrayList<>(); - - int remainingQuota = maxTotalBatchSize; - remainingQuota -= drainQueue(connectionAttemptQueue, connectionAttempts, remainingQuota); - remainingQuota -= drainQueue(statementExecutionQueue, statementExecutions, remainingQuota); - remainingQuota -= drainQueue(errorMetricQueue, errorMetrics, remainingQuota); - remainingQuota -= drainQueue(featureUsageQueue, featureUsages, remainingQuota); - - if (connectionAttempts.isEmpty() - && statementExecutions.isEmpty() - && errorMetrics.isEmpty() - && featureUsages.isEmpty()) { - return TransportResult.disabled(); - } + int maxTotalBatchSize = + config != null + ? config.getBatchSizeThreshold() + : TelemetryConfiguration.DEFAULT_BATCH_SIZE_THRESHOLD; + + List connectionAttempts = new ArrayList<>(); + List statementExecutions = new ArrayList<>(); + List errorMetrics = new ArrayList<>(); + List featureUsages = new ArrayList<>(); + + int remainingQuota = maxTotalBatchSize; + remainingQuota -= drainQueue(connectionAttemptQueue, connectionAttempts, remainingQuota); + remainingQuota -= drainQueue(statementExecutionQueue, statementExecutions, remainingQuota); + remainingQuota -= drainQueue(errorMetricQueue, errorMetrics, remainingQuota); + remainingQuota -= drainQueue(featureUsageQueue, featureUsages, remainingQuota); + + if (connectionAttempts.isEmpty() + && statementExecutions.isEmpty() + && errorMetrics.isEmpty() + && featureUsages.isEmpty()) { + return TransportResult.disabled(); + } - Instant now = Instant.now(); - Timestamp timestamp = - Timestamp.newBuilder().setSeconds(now.getEpochSecond()).setNanos(now.getNano()).build(); + Instant now = Instant.now(); + Timestamp timestamp = + Timestamp.newBuilder().setSeconds(now.getEpochSecond()).setNanos(now.getNano()).build(); - TelemetryPayload.Builder payloadBuilder = TelemetryPayload.newBuilder().setEventTime(timestamp); - if (driverEnvironment != null) { - payloadBuilder.setDriverEnvironment(driverEnvironment); - } - payloadBuilder.addAllConnectionAttempts(connectionAttempts); - payloadBuilder.addAllStatementExecutions(statementExecutions); - payloadBuilder.addAllErrors(errorMetrics); - payloadBuilder.addAllFeatureUsages(featureUsages); - - TelemetryPayload payload = payloadBuilder.build(); - - // Enforce MAX_PAYLOAD_BYTES (512 KB) by trimming excess items if needed - while (payload.getSerializedSize() > MAX_PAYLOAD_BYTES) { - if (!featureUsages.isEmpty()) { - featureUsageQueue.offer(featureUsages.remove(featureUsages.size() - 1)); - payloadBuilder.clearFeatureUsages().addAllFeatureUsages(featureUsages); - } else if (!errorMetrics.isEmpty()) { - errorMetricQueue.offer(errorMetrics.remove(errorMetrics.size() - 1)); - payloadBuilder.clearErrors().addAllErrors(errorMetrics); - } else if (!statementExecutions.isEmpty()) { - statementExecutionQueue.offer(statementExecutions.remove(statementExecutions.size() - 1)); - payloadBuilder.clearStatementExecutions().addAllStatementExecutions(statementExecutions); - } else if (!connectionAttempts.isEmpty()) { - connectionAttemptQueue.offer(connectionAttempts.remove(connectionAttempts.size() - 1)); - payloadBuilder.clearConnectionAttempts().addAllConnectionAttempts(connectionAttempts); - } else { - break; + TelemetryPayload.Builder payloadBuilder = + TelemetryPayload.newBuilder().setEventTime(timestamp); + if (driverEnvironment != null) { + payloadBuilder.setDriverEnvironment(driverEnvironment); + } + payloadBuilder.addAllConnectionAttempts(connectionAttempts); + payloadBuilder.addAllStatementExecutions(statementExecutions); + payloadBuilder.addAllErrors(errorMetrics); + payloadBuilder.addAllFeatureUsages(featureUsages); + + TelemetryPayload payload = payloadBuilder.build(); + + // Enforce MAX_PAYLOAD_BYTES (512 KB) by estimating bulk trim with safety padding + int currentSize = payload.getSerializedSize(); + if (currentSize > MAX_PAYLOAD_BYTES) { + int totalItems = + connectionAttempts.size() + + statementExecutions.size() + + errorMetrics.size() + + featureUsages.size(); + if (totalItems > 0) { + double avgBytesPerItem = (double) currentSize / totalItems; + int excessBytes = currentSize - MAX_PAYLOAD_BYTES; + // Add +5 items safety pad to guarantee 100% single-pass size compliance + int itemsToTrim = + Math.min(totalItems, (int) Math.ceil(excessBytes / avgBytesPerItem) + 5); + + trimExcessItemsBulk( + itemsToTrim, connectionAttempts, statementExecutions, errorMetrics, featureUsages); + + payloadBuilder + .clearConnectionAttempts() + .addAllConnectionAttempts(connectionAttempts) + .clearStatementExecutions() + .addAllStatementExecutions(statementExecutions) + .clearErrors() + .addAllErrors(errorMetrics) + .clearFeatureUsages() + .addAllFeatureUsages(featureUsages); + payload = payloadBuilder.build(); + } } - payload = payloadBuilder.build(); - } - TransportResult result; - try { - result = transport.send(payload); - } catch (Throwable t) { - logger.log(Level.WARNING, "Unexpected exception during telemetry flush", t); - result = new TransportResult(false, -1); - } + TransportResult result; + try { + result = transport.send(payload); + } catch (Throwable t) { + logger.log(Level.WARNING, "Unexpected exception during telemetry flush", t); + result = new TransportResult(false, -1); + } - if (!result.isSuccess()) { - requeueItems(connectionAttemptQueue, connectionAttempts); - requeueItems(statementExecutionQueue, statementExecutions); - requeueItems(errorMetricQueue, errorMetrics); - requeueItems(featureUsageQueue, featureUsages); - } + if (!result.isSuccess()) { + requeueItems(connectionAttemptQueue, connectionAttempts); + requeueItems(statementExecutionQueue, statementExecutions); + requeueItems(errorMetricQueue, errorMetrics); + requeueItems(featureUsageQueue, featureUsages); + } - long uploadIntervalMs = config != null ? config.getUploadIntervalMs() : 300_000L; - long newDelayMs = - result.getNextRequestWaitMillis() > 0 - ? Math.max(uploadIntervalMs, result.getNextRequestWaitMillis()) - : uploadIntervalMs; - reschedule(newDelayMs); + long uploadIntervalMs = config != null ? config.getUploadIntervalMs() : 300_000L; + long newDelayMs = + result.getNextRequestWaitMillis() > 0 + ? Math.max(uploadIntervalMs, result.getNextRequestWaitMillis()) + : uploadIntervalMs; + reschedule(newDelayMs); - return result; + return result; + } finally { + flushLock.unlock(); + } } @Override @@ -211,10 +230,13 @@ public void close() { return; } - synchronized (this) { + flushLock.lock(); + try { if (scheduledTask != null) { scheduledTask.cancel(false); } + } finally { + flushLock.unlock(); } try { @@ -254,34 +276,66 @@ private boolean isConfigured() { return config != null && config.isEnabled() && transport != null; } - private synchronized void reschedule(long delayMs) { + private void reschedule(long delayMs) { if (isClosed.get() || executorService == null || executorService.isShutdown()) { return; } long effectiveDelay = Math.max(1L, delayMs); - if (currentScheduleDelayMs.get() == effectiveDelay - && scheduledTask != null - && !scheduledTask.isDone()) { - return; - } - if (scheduledTask != null) { - scheduledTask.cancel(false); + flushLock.lock(); + try { + if (currentScheduleDelayMs.get() == effectiveDelay + && scheduledTask != null + && !scheduledTask.isDone()) { + return; + } + + if (scheduledTask != null) { + scheduledTask.cancel(false); + } + + currentScheduleDelayMs.set(effectiveDelay); + scheduledTask = + executorService.scheduleWithFixedDelay( + () -> { + try { + flush(); + } catch (Throwable t) { + logger.log(Level.WARNING, "Periodic telemetry flush encountered an exception", t); + } + }, + effectiveDelay, + effectiveDelay, + TimeUnit.MILLISECONDS); + } finally { + flushLock.unlock(); } + } - currentScheduleDelayMs.set(effectiveDelay); - scheduledTask = - executorService.scheduleWithFixedDelay( - () -> { - try { - flush(); - } catch (Throwable t) { - logger.log(Level.WARNING, "Periodic telemetry flush encountered an exception", t); - } - }, - effectiveDelay, - effectiveDelay, - TimeUnit.MILLISECONDS); + private void trimExcessItemsBulk( + int itemsToTrim, + List connectionAttempts, + List statementExecutions, + List errorMetrics, + List featureUsages) { + int remainingToTrim = itemsToTrim; + + while (remainingToTrim > 0 && !featureUsages.isEmpty()) { + featureUsageQueue.offer(featureUsages.remove(featureUsages.size() - 1)); + remainingToTrim--; + } + while (remainingToTrim > 0 && !errorMetrics.isEmpty()) { + errorMetricQueue.offer(errorMetrics.remove(errorMetrics.size() - 1)); + remainingToTrim--; + } + while (remainingToTrim > 0 && !statementExecutions.isEmpty()) { + statementExecutionQueue.offer(statementExecutions.remove(statementExecutions.size() - 1)); + remainingToTrim--; + } + while (remainingToTrim > 0 && !connectionAttempts.isEmpty()) { + connectionAttemptQueue.offer(connectionAttempts.remove(connectionAttempts.size() - 1)); + remainingToTrim--; + } } private int drainQueue(ConcurrentLinkedQueue queue, List targetList, int maxItems) { From 82483ba107cb0f5f89cefed17b6eddf62fcb8947 Mon Sep 17 00:00:00 2001 From: Neenu1995 Date: Tue, 14 Jul 2026 17:21:38 -0400 Subject: [PATCH 3/5] use queue --- .../bigquery/jdbc/telemetry/v1/TelemetryBatcher.java | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryBatcher.java b/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryBatcher.java index cd65b9e899a7..682ab8b96663 100644 --- a/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryBatcher.java +++ b/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryBatcher.java @@ -21,6 +21,7 @@ import java.time.Instant; import java.util.ArrayList; import java.util.List; +import java.util.Queue; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; @@ -338,7 +339,7 @@ private void trimExcessItemsBulk( } } - private int drainQueue(ConcurrentLinkedQueue queue, List targetList, int maxItems) { + private int drainQueue(Queue queue, List targetList, int maxItems) { int count = 0; T item; while (count < maxItems && (item = queue.poll()) != null) { @@ -348,11 +349,9 @@ private int drainQueue(ConcurrentLinkedQueue queue, List targetList, i return count; } - private void requeueItems(ConcurrentLinkedQueue queue, List items) { - for (T item : items) { - if (queue.size() < MAX_QUEUE_SIZE) { - queue.offer(item); - } + private void requeueItems(Queue queue, List items) { + if (items != null && !items.isEmpty()) { + queue.addAll(items); } } } From 6393b06b0cf116a68ffe44aab699c34947e77934 Mon Sep 17 00:00:00 2001 From: Neenu1995 Date: Tue, 14 Jul 2026 17:30:01 -0400 Subject: [PATCH 4/5] use blocking queue --- .../jdbc/telemetry/v1/TelemetryBatcher.java | 21 ++++++++----------- 1 file changed, 9 insertions(+), 12 deletions(-) diff --git a/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryBatcher.java b/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryBatcher.java index 682ab8b96663..e98dde3764c9 100644 --- a/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryBatcher.java +++ b/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryBatcher.java @@ -22,8 +22,8 @@ import java.util.ArrayList; import java.util.List; import java.util.Queue; -import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.Executors; +import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; @@ -51,13 +51,12 @@ final class TelemetryBatcher implements AutoCloseable { private final boolean ownsExecutor; private final ReentrantLock flushLock = new ReentrantLock(); - private final ConcurrentLinkedQueue connectionAttemptQueue = - new ConcurrentLinkedQueue<>(); - private final ConcurrentLinkedQueue statementExecutionQueue = - new ConcurrentLinkedQueue<>(); - private final ConcurrentLinkedQueue errorMetricQueue = new ConcurrentLinkedQueue<>(); - private final ConcurrentLinkedQueue featureUsageQueue = - new ConcurrentLinkedQueue<>(); + private final Queue connectionAttemptQueue = + new LinkedBlockingQueue<>(MAX_QUEUE_SIZE); + private final Queue statementExecutionQueue = + new LinkedBlockingQueue<>(MAX_QUEUE_SIZE); + private final Queue errorMetricQueue = new LinkedBlockingQueue<>(MAX_QUEUE_SIZE); + private final Queue featureUsageQueue = new LinkedBlockingQueue<>(MAX_QUEUE_SIZE); private final AtomicBoolean isClosed = new AtomicBoolean(false); private final AtomicLong currentScheduleDelayMs = new AtomicLong(-1); @@ -109,13 +108,11 @@ void offerFeatureUsage(FeatureUsage featureUsage) { offerMetric(featureUsageQueue, featureUsage, "Feature usage"); } - private void offerMetric(ConcurrentLinkedQueue queue, T item, String metricName) { + private void offerMetric(Queue queue, T item, String metricName) { if (isClosed.get() || !isConfigured() || item == null) { return; } - if (queue.size() < MAX_QUEUE_SIZE) { - queue.offer(item); - } else { + if (!queue.offer(item)) { logger.log(Level.WARNING, metricName + " telemetry queue full. Dropping metric."); } } From 144ef4617a9bad54940cb6bf2affd561e5268d3a Mon Sep 17 00:00:00 2001 From: Neenu1995 Date: Wed, 15 Jul 2026 12:16:04 -0400 Subject: [PATCH 5/5] address gemini comments --- .../jdbc/telemetry/v1/TelemetryBatcher.java | 73 ++++++++++--------- 1 file changed, 38 insertions(+), 35 deletions(-) diff --git a/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryBatcher.java b/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryBatcher.java index e98dde3764c9..42256f6f52c7 100644 --- a/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryBatcher.java +++ b/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/telemetry/v1/TelemetryBatcher.java @@ -51,12 +51,14 @@ final class TelemetryBatcher implements AutoCloseable { private final boolean ownsExecutor; private final ReentrantLock flushLock = new ReentrantLock(); - private final Queue connectionAttemptQueue = + private final LinkedBlockingQueue connectionAttemptQueue = new LinkedBlockingQueue<>(MAX_QUEUE_SIZE); - private final Queue statementExecutionQueue = + private final LinkedBlockingQueue statementExecutionQueue = + new LinkedBlockingQueue<>(MAX_QUEUE_SIZE); + private final LinkedBlockingQueue errorMetricQueue = + new LinkedBlockingQueue<>(MAX_QUEUE_SIZE); + private final LinkedBlockingQueue featureUsageQueue = new LinkedBlockingQueue<>(MAX_QUEUE_SIZE); - private final Queue errorMetricQueue = new LinkedBlockingQueue<>(MAX_QUEUE_SIZE); - private final Queue featureUsageQueue = new LinkedBlockingQueue<>(MAX_QUEUE_SIZE); private final AtomicBoolean isClosed = new AtomicBoolean(false); private final AtomicLong currentScheduleDelayMs = new AtomicLong(-1); @@ -66,13 +68,17 @@ final class TelemetryBatcher implements AutoCloseable { this( config, transport, - Executors.newSingleThreadScheduledExecutor( - r -> { - Thread t = new Thread(r, "jdbc-telemetry-batcher"); - t.setDaemon(true); - return t; - }), - true); + (config != null && config.isEnabled()) ? createDefaultExecutor() : null, + config != null && config.isEnabled()); + } + + private static ScheduledExecutorService createDefaultExecutor() { + return Executors.newSingleThreadScheduledExecutor( + r -> { + Thread t = new Thread(r, "jdbc-telemetry-batcher"); + t.setDaemon(true); + return t; + }); } // Package-private constructor for testing overrides @@ -316,39 +322,36 @@ private void trimExcessItemsBulk( List statementExecutions, List errorMetrics, List featureUsages) { - int remainingToTrim = itemsToTrim; + int remaining = itemsToTrim; + remaining -= trimListToQueue(featureUsages, featureUsageQueue, remaining); + remaining -= trimListToQueue(errorMetrics, errorMetricQueue, remaining); + remaining -= trimListToQueue(statementExecutions, statementExecutionQueue, remaining); + trimListToQueue(connectionAttempts, connectionAttemptQueue, remaining); + } - while (remainingToTrim > 0 && !featureUsages.isEmpty()) { - featureUsageQueue.offer(featureUsages.remove(featureUsages.size() - 1)); - remainingToTrim--; - } - while (remainingToTrim > 0 && !errorMetrics.isEmpty()) { - errorMetricQueue.offer(errorMetrics.remove(errorMetrics.size() - 1)); - remainingToTrim--; - } - while (remainingToTrim > 0 && !statementExecutions.isEmpty()) { - statementExecutionQueue.offer(statementExecutions.remove(statementExecutions.size() - 1)); - remainingToTrim--; - } - while (remainingToTrim > 0 && !connectionAttempts.isEmpty()) { - connectionAttemptQueue.offer(connectionAttempts.remove(connectionAttempts.size() - 1)); - remainingToTrim--; + private int trimListToQueue(List list, Queue queue, int maxToTrim) { + int trimmed = 0; + while (trimmed < maxToTrim && !list.isEmpty()) { + queue.offer(list.remove(list.size() - 1)); + trimmed++; } + return trimmed; } - private int drainQueue(Queue queue, List targetList, int maxItems) { - int count = 0; - T item; - while (count < maxItems && (item = queue.poll()) != null) { - targetList.add(item); - count++; + private int drainQueue(LinkedBlockingQueue queue, List targetList, int maxItems) { + if (maxItems <= 0) { + return 0; } - return count; + return queue.drainTo(targetList, maxItems); } private void requeueItems(Queue queue, List items) { if (items != null && !items.isEmpty()) { - queue.addAll(items); + for (T item : items) { + if (!queue.offer(item)) { + break; // Queue is full; stop requeueing to avoid dropping metrics or throwing exception + } + } } } }