sebastienviale commented on code in PR #22725:
URL: https://github.com/apache/kafka/pull/22725#discussion_r3614855475
##########
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:
The 0 case comes from the fact that a store name may not be registered in
any sub-topology. In that case, partitionsOf() returns 0 rather than throwing.
I will clarify the Javadoc.
##########
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).
+ * @throws IllegalStateException if the driver is not operating in
multi-partition mode
+ */
+ int partitionsOf(final String storeName) {
+ requireMultiPartitionMode();
+ return runtime.partitionsOf(storeName);
+ }
+
+ /**
+ * @return the number of partitions of the given sub-topology, or 0 if the
id is unknown.
Review Comment:
Good point. Since 0 is not a valid partition count, returning it as a
fallback could hide an invalid sub-topology id. I will change it to throw an
exception.
--
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]
