mjsax commented on code in PR #22725:
URL: https://github.com/apache/kafka/pull/22725#discussion_r3618651235
##########
streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java:
##########
@@ -979,7 +1118,91 @@ private StateStore getStateStore(final String name,
return null;
}
- private void throwIfBuiltInStore(final StateStore stateStore) {
+ /**
+ * Return the {@link StateStore} for the task owning {@code partition} of
the sub-topology that
+ * registers a store named {@code name}. If the store name appears in
multiple
+ * sub-topologies, throws {@link IllegalStateException}.
+ *
+ * @param name the store name
+ * @param partition the partition whose owning task should be queried
+ * @return the {@link StateStore}, or {@code null} if no sub-topology
registers a store with this name
+ */
+ public StateStore getStateStore(final String name, final int partition) {
+ requireMultiPartitionMode();
+ return runtime.getStateStore(name, partition);
+ }
+
+ /**
+ * Guard for the partition-aware accessors below ({@link
#getStateStore(String, int)} and
+ * friends). Unlike the original implementation, this never activates
multi-partition mode as a
+ * side effect of what looks like a read-only getter: a {@code getXxx()}
method that can silently
+ * rebuild the entire task graph on first call -- and behave differently
the second time it's
+ * called -- is surprising and hides a non-trivial effect behind an
innocuous-looking signature.
+ *
+ * <p>Multi-partition mode must already be active by the time these
accessors are called, either
+ * because {@link TopologyTestDriverBuilder#build()} activated it (at
least one declared topic has
+ * more than one partition), or because the caller invoked {@link
#activateMultiPartitionMode()}
+ * explicitly. If it isn't, this throws -- it does not activate it for you.
+ *
+ * @throws IllegalStateException if the driver is not operating in
multi-partition mode
+ */
+ private void requireMultiPartitionMode() {
+ if (!multiPartitionModeActive) {
+ throw new IllegalStateException(
+ "This driver is not operating in multi-partition mode. Declare
a topic with more than "
+ + "one partition (via
TopologyTestDriverBuilder#declareTopic() or declareTopic()) "
+ + "and call activateMultiPartitionMode() -- or simply pipe
a record first, which "
+ + "activates it automatically -- before calling
partition-aware accessors like "
+ + "getStateStore(name, partition). Use getStateStore(name)
for single-partition mode.");
+ }
+ }
+
+ /**
+ * Internal fully-qualified {@link StateStore} accessor: resolves a store
to the task owning
+ * {@code (subtopologyId, partition)}. Package-private -- not part of the
KIP-1238 public API;
+ * callers use {@link #getStateStore(String, int)}, which resolves the
sub-topology by store name.
+ *
+ * @param name the store name
+ * @param subtopologyId the sub-topology id
+ * @param partition the partition whose owning task should be queried
+ * @return the {@link StateStore}, or {@code null} if the task does not
register a store with this name
+ * @throws IllegalArgumentException if no task exists for {@code
(subtopologyId, partition)}
+ * @throws IllegalStateException if the driver is not operating in
multi-partition mode
+ */
+ StateStore getStateStore(final String name, final int subtopologyId, final
int partition) {
+ requireMultiPartitionMode();
+ return runtime.getStateStore(name, subtopologyId, partition);
+ }
+
+ /**
+ * @return the number of partitions of the sub-topology that registers
{@code storeName}, or 0
+ * if no sub-topology registers it (or 1 for a global store).
Review Comment:
Hmmm... So you mean like a "incorrectly" added store, that was never
attached to any `Processor` (aka "dangling store")? -- Such a store should
actually just disappear at runtime, at least this happens in the real KS
runtime. Not sure if TDD would drop it -- If TTD does not drop it, it maybe
should? Or is this not possible?
Wondering if returning partition 0 for such a store would be correct, of if
TDD should just throw instead?
Or do you mean something else?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
