sebastienviale commented on code in PR #22725:
URL: https://github.com/apache/kafka/pull/22725#discussion_r3614849684
##########
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.
Review Comment:
Good point
##########
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 "
Review Comment:
done
--
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]
