feat(bigquery-jdbc): add aggregation logic

[Neenu1995]

No description provided.

[gemini-code-assist]

Code Review

This pull request refactors the telemetry batching mechanism in TelemetryBatcher and TelemetryManager to use lock-free, in-memory accumulators (ConcurrentHashMap and LongAdder) instead of a blocking queue of Protobuf messages. This optimization eliminates object allocation and GC overhead during telemetry recording. The reviewer identified a potential issue in the refactored reschedule method, which is missing an isClosed guard, potentially leading to a RejectedExecutionException or scheduling tasks on a closed batcher during shutdown.

[gemini-code-assist]

Missing isClosed Guard in reschedule

The refactored reschedule method is missing the check for isClosed.get().

When close() is called, it executes a final flush(). This final flush() will call reschedule(), which will attempt to schedule another task on the executorService. Since close() is also shutting down the executor service, this can throw a RejectedExecutionException or lead to tasks being scheduled on a closed batcher.

We should restore the isClosed.get() check at the beginning of reschedule.

Suggested change

	private void reschedule(long delayMs) {
	long current = currentScheduleDelayMs.get();
	if (current == delayMs && scheduledTask != null && !scheduledTask.isDone()) {
	return;
	}
	for (ErrorMetric.Builder b : errors.values()) {
	builder.addErrors(b);
	if (scheduledTask != null) {
	scheduledTask.cancel(false);
	}
	for (FeatureUsage.Builder b : features.values()) {
	builder.addFeatureUsages(b);
	if (executorService != null && !executorService.isShutdown()) {
	currentScheduleDelayMs.set(delayMs);
	scheduledTask = executorService.schedule(this::flush, delayMs, TimeUnit.MILLISECONDS);
	}
	}
	private void reschedule(long delayMs) {
	if (isClosed.get() || executorService == null || executorService.isShutdown()) {
	return;
	}
	long current = currentScheduleDelayMs.get();
	if (current == delayMs && scheduledTask != null && !scheduledTask.isDone()) {
	return;
	}
	if (scheduledTask != null) {
	scheduledTask.cancel(false);
	}
	currentScheduleDelayMs.set(delayMs);
	scheduledTask = executorService.schedule(this::flush, delayMs, TimeUnit.MILLISECONDS);
	}

References

1. When using lazily initialized resources (such as ExecutorService), ensure that teardown or close methods perform explicit null checks before invoking methods on them to prevent NullPointerException.