KSQL-14980 | Address misc items for ksql

[Parag Badani (pbadani)]

Description

What behavior do you want to change, why, how does your patch achieve the changes?
Address misc items for ksql

Testing done

Describe the testing strategy. Unit and integration tests are expected for any behavior changes.

Reviewer checklist

• Ensure docs are updated if necessary. (eg. if a user visible feature is being added or changed).
• Ensure relevant issues are linked (description should include text like "Fixes #")
• Do these changes have compatibility implications for rollback? If so, ensure that the ksql command version is bumped.

[confluent-cla-assistant]

🎉 All Contributor License Agreements have been signed. Ready to merge.
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}

[Vedarth Sharma (VedarthConfluent)]

Code review

Found 1 issue:

1. Duplicate import java.util.Map in QueryRegistryImplTest.java — the PR inserts a second import java.util.Map; at line 48, but the original file already has one at line 52. The post-merge file has both, which is a Checkstyle violation (the build enforces CustomImportOrder with no duplicates). This will fail CI.

ksql/ksqldb-engine/src/test/java/io/confluent/ksql/query/QueryRegistryImplTest.java

Lines 46 to 55 in fddf15d

	import java.lang.reflect.Field;
	import java.util.Arrays;
	import java.util.Map;
	import java.util.Collection;
	import java.util.HashMap;
	import java.util.List;
	import java.util.Map;
	import java.util.Optional;
	import java.util.OptionalInt;
	import java.util.Set;

Fix: remove the newly added import java.util.Map; at line 48 (or consolidate so only one remains), and also add the missing import java.util.List; in the correct sorted position.

---

Other changes reviewed and look correct (historical context checked via git log and git blame):

• BinPackedPersistentQueryMetadataImpl.resume() calling listener.onResume instead of onPause — genuine bug fix.
• StorageUtilizationMetricsReporter decrement-after-guard fix — correct; prevents negative num_stateful_tasks for multi-store tasks.
• TransientQueryCleanupService.isLeaked reordering — correct; live-query short-circuit must run first to prevent deletion of active topics.
• DistributingExecutor rate-limit ordering + producer close on initTransactions failure — correct; the old ordering leaked the producer on every throttled request.
• All volatile additions, CopyOnWriteArrayList, ConcurrentHashMap/putIfAbsent, and interrupt-restore in RetryUtil look correct and are consistent with historical threading fixes in the same files.

🤖 Generated with Claude Code

_{- If this code review was useful, please react with 👍. Otherwise, react with 👎.}

[Vedarth Sharma (VedarthConfluent)]

Code review

Found 3 issues:

1. SharedKafkaStreamsRuntimeImpl.stop() iterates topolgogiesToAdd (now a Collections.synchronizedList) and then calls .clear() without holding the list's intrinsic lock. start() concurrently appends via topolgogiesToAdd.add(...). Per the Collections.synchronizedList contract, iteration requires external synchronized(list) { ... } or a ConcurrentModificationException may be thrown. restartStreamsRuntime() in the same PR correctly wraps the clear() in synchronized (topolgogiesToAdd) { ... }; stop() should do the same around the for-each + clear.

ksql/ksqldb-engine/src/main/java/io/confluent/ksql/util/SharedKafkaStreamsRuntimeImpl.java

Lines 193 to 200 in fddf15d

	if (kafkaStreams.state().isRunningOrRebalancing()) {
	try {
	for (KafkaFuture<Void> toAdd : topolgogiesToAdd) {
	toAdd.get();
	}
	topolgogiesToAdd.clear();
	kafkaStreams.removeNamedTopology(queryId.toString(), resetOffsets)
	.all()

2. QueryMetadataImpl.TimeBoundedQueue.toImmutableList() calls ImmutableList.copyOf(queue) on a Queues.synchronizedQueue(...) without external synchronization. The Guava/JDK contract for synchronized collection wrappers requires manual synchronized(queue) { ... } while iterating; ImmutableList.copyOf uses the queue's iterator. queryErrors.add() runs on the Streams uncaught-handler thread while getQueryErrors()→toImmutableList() runs from REST handlers, so this can race and throw CME. The comment at line 391 ("all queue operations are thread-safe under concurrent add/evict calls") overstates what synchronizedQueue provides for iteration.

ksql/ksqldb-engine/src/main/java/io/confluent/ksql/util/QueryMetadataImpl.java

Lines 404 to 410 in fddf15d

	}
	public List<QueryError> toImmutableList() {
	evict();
	return ImmutableList.copyOf(queue);
	}

3. TransientQueryQueue.limitHandler is written by the REST thread in setLimitHandler() and read by the Streams thread in onQueued(), but the field is not volatile. The added null-check (final LimitHandler handler = limitHandler; if (handler != null) ...) prevents the NPE but does not establish a happens-before relationship — the Streams thread may continue to observe a stale null indefinitely under the Java Memory Model. The passedLimit() fallback in setLimitHandler() mitigates the user-visible symptom, but the field should be volatile to be correct (matching private volatile boolean closed = false; on line 43 of the same class, which handles an analogous cross-thread access).

ksql/ksqldb-engine/src/main/java/io/confluent/ksql/query/TransientQueryQueue.java

Lines 43 to 46 in fddf15d

	private volatile boolean closed = false;
	private final AtomicInteger remaining;
	private LimitHandler limitHandler;
	private CompletionHandler completionHandler;

ksql/ksqldb-engine/src/main/java/io/confluent/ksql/query/TransientQueryQueue.java

Lines 188 to 194 in fddf15d

	// passedLimit() block fires limitReached once the handler is finally
	// installed, so dropping it here is safe.
	final LimitHandler handler = limitHandler;
	if (handler != null) {
	handler.limitReached();
	}
	}

🤖 Generated with Claude Code

_{- If this code review was useful, please react with 👍. Otherwise, react with 👎.}