KAFKA-13491: IQv2 framework #11557

vvcephei · 2021-11-30T22:23:26Z

Implements a new interactive query framework,
as described in KIP-805 (https://cwiki.apache.org/confluence/x/85OqCw).

No public queries are added in this PR, just the framework and tests.

Also, position tracking and bounding is not implemented in this PR.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

vvcephei

Hey @patrickstuedi and @vpapavas, I'd really appreciate your reviews on this. Also, @cadonna , @bbejeck , or @guozhangwang , since you were involved in the discussion and voting, would you mind giving this a look?

I'm sorry about the length, but it's hopefully a pretty straightforward implementation of the KIP, with a lot of repetition and boilerplate.

Thanks in advance for your time!

vvcephei · 2021-11-30T22:23:46Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

+     *                                    topology.
+     */
+    @Evolving
+    public <R> StateQueryResult<R> query(final StateQueryRequest<R> request) {


Main entry point for the new API.

This implementation may look inefficient, but it's actually identical to the way that interactive queries are currently served. I think it could be improved, but I'm leaving it to future PRs.

vvcephei · 2021-11-30T22:25:40Z

streams/src/main/java/org/apache/kafka/streams/errors/StreamsStoppedException.java

+ * State#ERROR}. This Streams instance will need to be discarded and replaced before it can
+ * serve queries. The caller may wish to query a different instance.
+ */
+public class StreamsStoppedException extends InvalidStateStoreException {


New exception proposed in the KIP. See KafkaStreams#query javadoc.

vvcephei · 2021-11-30T22:26:48Z

streams/src/main/java/org/apache/kafka/streams/processor/StateStoreContext.java

+     * This is defined as the metadata of the record that is currently being
+     * processed (or was last processed) by the StreamTask that holds the store.


This is a bit of a drive-by; I just noticed the wording could be improved.

vvcephei · 2021-11-30T22:28:15Z

streams/src/main/java/org/apache/kafka/streams/query/Position.java

+ * reference.
+ */
+@Evolving
+public class Position {


This adds the public Position API that was proposed in the KIP. I've also gone ahead and removed the private Position api (from the state.internals package).

vvcephei · 2021-11-30T22:32:00Z

streams/src/main/java/org/apache/kafka/streams/state/Stores.java

-                return "in-memory";
-            }
-        };
+        return new InMemoryKeyValueBytesStoreSupplier(name);


Having this inline made it harder to debug my test, so I just popped it out into a separate class.

vvcephei · 2021-11-30T23:15:15Z

streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java

+     * Once position bounding is generally supported, we should migrate tests to wait on the
+     * expected response position.
+     */
+    public static <R> StateQueryResult<R> iqv2WaitForPartitionsOrGlobal(


The javadoc pretty much sums this up. It's a harbinger of good things to come, since as you can see, we'll no longer be vulnerable to race conditions like rebalances sneaking in during the test. We can just wait for a valid result and then verify it.

vvcephei · 2021-11-30T23:15:42Z

streams/src/test/java/org/apache/kafka/streams/query/PositionBoundTest.java

+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+public class PositionBoundTest {


Just a basic unit test.

vvcephei · 2021-11-30T23:16:12Z

streams/src/test/java/org/apache/kafka/streams/query/PositionTest.java

+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertThrows;
+
+public class PositionTest {


Another basic unit test, based on the internal API one that I removed.

vvcephei · 2021-11-30T23:17:10Z

streams/src/test/java/org/apache/kafka/streams/query/PositionTest.java

+    }
+
+    @Test
+    public void shouldNotHash() {


note: since this class is mutable, I wanted to make sure we don't accidentally use it in sets or anything.

vvcephei · 2021-11-30T23:17:34Z

streams/src/test/java/org/apache/kafka/streams/state/internals/PositionTest.java

-import java.io.IOException;
-import org.junit.Test;
-
-public class PositionTest {


These cases are migrated into the new unit test.

patrickstuedi · 2021-12-01T14:10:32Z

streams/src/main/java/org/apache/kafka/streams/query/Position.java

+            .computeIfAbsent(topic, k -> new ConcurrentHashMap<>())
+            .compute(
+                partition,
+                (integer, prior) -> prior == null || offset > prior ? offset : prior


Nice, this avoids the use of AtomicLong!

patrickstuedi · 2021-12-01T14:11:22Z

streams/src/main/java/org/apache/kafka/streams/query/Position.java

+     * Create a deep copy of the Position.
+     */
+    public Position copy() {
+        return new Position(deepCopy(position));


Why not implement copy using merge?
Position pos = Position.empyPosition();
pos.merge(this);

Thanks for the review!

That would also work, but this seems more direct.

vpapavas · 2021-12-01T14:23:49Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

+                                || !active)) {
+                                result.addResult(
+                                    partition,
+                                    QueryResult.forFailure(


Why do we add a failure here? The for loop will continue iterating and it might find an active task that hosts that store partition later, right? Don't you want to add the failure at the end of the loop if no such active partition is found?
Moreover, don't you want to exit the loop after the failure?

Also, the failure reason is a bit misleading

Thanks for the review!

We can just add the failure right away because tasks are uniquely assigned to Streams instances, so it's not possible to find a single task twice in this loop.

That's a good point about exiting as soon at the query is complete. I've updated the PR.

Also, that's a good point about the failure reason. I previously had a specific PositionBound to "requireLatest" (aka "require active"), but I moved it into the StateQueryRequest, since enforcing the task state is the responsibility of the framework, not a store. However, I never changed this FailureReason. I've fixed it with a new reason in the PR. Good catch!

vvcephei · 2021-12-01T18:11:59Z

Rebased to resolve conflicts with #11541

patrickstuedi · 2021-12-03T09:23:43Z

streams/src/main/java/org/apache/kafka/streams/state/internals/StoreQueryUtils.java

+        final long start = collectExecutionInfo ? System.nanoTime() : -1L;
+        // TODO: position tracking
+        if (query instanceof PingQuery) {
+            result = (QueryResult<R>) QueryResult.forResult(true);


How do we ensure type R matches the type of template type of the store?

Unfortunately, it's not possible to guarantee that using Java's type system. The query caller will just have to be sure to submit correctly typed queries based on their a priori knowledge of the store's types. If they get it wrong, they'll get a ClassCastException at run time. This is a bummer, but it's also precisely the same in IQ today.

As far as I know, it's not possible to do better than this.

cadonna

@vvcephei , Thanks for the PR!

You probably mean KIP-796 in the description, right?

I will approve your PR since it blocks other PR and it doesn't seem to be risky to merge it. Please consider my feedback for a follow-up PR.

cadonna · 2021-12-03T09:10:31Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

+        final Map<String, StateStore> globalStateStores = topologyMetadata.globalStateStores();
+        if (globalStateStores.containsKey(storeName)) {
+            final StateStore store = globalStateStores.get(storeName);
+            final QueryResult<R> r =


nit: Could you give a more meaningful name to this variable?

Sure! Sorry about that.

cadonna · 2021-12-03T09:22:06Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

+                        if (!request.isAllPartitions()
+                            && handledPartitions.containsAll(request.getPartitions())) {
+                            return result;
+                        }


nit: Instead of putting a comment, could you extract the condition and give it a meaningful name?

Ah, I see what you mean. Yes, it's a bit mysterious this way.

cadonna · 2021-12-03T09:25:29Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

+        }
+
+        return result;
+    }


In general, the method is quite long. Could you try to extract methods with meaningful names? For example, the checks in the beginning to verify whether the state is known or the stream thread is running can be extracted to methods.

For sure! I was actually hoping that we would have a chance to make this logic more efficient, so I didn't bother making it too pretty, but you're totally right. There's no guarantee on when or if we'll refactor it, so we should make it nice now.

cadonna · 2021-12-03T09:26:18Z

checkstyle/suppressions.xml

@@ -164,7 +164,7 @@
              files="(KafkaStreams|KStreamImpl|KTableImpl).java"/>

    <suppress checks="CyclomaticComplexity"
-              files="(StreamsPartitionAssignor|StreamThread|TaskManager|PartitionGroup).java"/>
+              files="(KafkaStreams|StreamsPartitionAssignor|StreamThread|TaskManager|PartitionGroup).java"/>


Is this really needed?

Yes, for the same reason that you thought the query method was too long and complicated :)

cadonna · 2021-12-03T09:34:24Z

streams/src/main/java/org/apache/kafka/streams/processor/StateStore.java

+        Query<R> query,
+        PositionBound positionBound,
+        boolean collectExecutionInfo) {
+        // If a store doesn't implement a query handler, then all queries are unknown.


Could you add this to the javadocs since it seems to be an important information for the outside world?

You're totally right. I thought I had said that already, but maybe I'm thinking of some other conversation.

cadonna · 2021-12-03T10:23:00Z

streams/src/main/java/org/apache/kafka/streams/query/QueryResult.java

+ *
+ * @param <R> The result type of the query.
+ */
+public final class QueryResult<R> {


Could you add unit tests for this class?

I actually decided not to for the classes you pointed out because they are all trivial "struct"-type classes with no logic. I did, however, verify that all the new code has 100% (or as close as possible) code coverage via the existing set of tests in this PR.

cadonna · 2021-12-03T10:33:52Z

streams/src/main/java/org/apache/kafka/streams/query/StateQueryResult.java

+     * The query's result for global store queries. Is {@code null} for non-global (partitioned)
+     * store queries.
+     */
+    public QueryResult<R> getGlobalResult() {


I could not find this method in the KIP. Could you update the KIP?

cadonna · 2021-12-03T10:35:00Z

streams/src/main/java/org/apache/kafka/streams/query/StateQueryResult.java

+    /**
+     * Set the result for a partitioned store query. Used by Kafka Streams and available for tests.
+     */
+    public void addResult(final int partition, final QueryResult<R> r) {


I could not find this method in the KIP. Could you update the KIP?

cadonna · 2021-12-03T10:35:05Z

streams/src/main/java/org/apache/kafka/streams/query/StateQueryResult.java

+    /**
+     * Set the result for a global store query. Used by Kafka Streams and available for tests.
+     */
+    public void setGlobalResult(final QueryResult<R> r) {


I could not find this method in the KIP. Could you update the KIP?

Gah! I'm sorry for that miss. I had it in the POC from the beginning, but overlooked it in the KIP.

cadonna · 2021-12-03T10:49:18Z

streams/src/main/java/org/apache/kafka/streams/state/internals/StoreQueryUtils.java

+import org.apache.kafka.streams.query.Query;
+import org.apache.kafka.streams.query.QueryResult;
+
+public final class StoreQueryUtils {


Could you add unit tests for this class?

Yeah, you're right. These do need unit tests.

vvcephei · 2021-12-03T18:42:26Z

Thanks again for the review, @cadonna ! I'm tempted to fix that stuff in the current PR, but I'd like to unblock others from rebasing their work on top of this framework. Since none of the feedback would invalidate those rebases, I'll go ahead and merge and then follow up asap with a PR to address your feedback.

Thanks so much for your time.

vvcephei · 2021-12-03T18:51:46Z

The failures are unrelated:

    Build / ARM / kafka.server.ReplicaManagerTest.[1] usesTopicIds=true
    Build / JDK 8 and Scala 2.12 / kafka.server.ReplicaManagerTest.[1] usesTopicIds=true
    Build / JDK 11 and Scala 2.13 / kafka.server.ReplicaManagerTest.[1] usesTopicIds=true

Implements the major part of the IQv2 framework as proposed in KIP-796. Reviewers: Patrick Stuedi <pstuedi@apache.org>, Vicky Papavasileiou <vpapavasileiou@confluent.io>, Bruno Cadonnna <cadonna@apache.org>

vvcephei commented Nov 30, 2021

View reviewed changes

vvcephei mentioned this pull request Dec 1, 2021

KAFKA-13498: keeping track of latest seen position in different state stores #11541

Merged

vvcephei added the streams label Dec 1, 2021

patrickstuedi reviewed Dec 1, 2021

View reviewed changes

vpapavas reviewed Dec 1, 2021

View reviewed changes

vvcephei force-pushed the iqv2-framework branch from dec3f37 to d531d4d Compare December 1, 2021 17:57

John Roesler added 2 commits December 1, 2021 11:58

IQv2 Framework

d531d4d

add position update util to avoid missed isPresent checks

27e689e

style

26604d6

patrickstuedi reviewed Dec 3, 2021

View reviewed changes

cadonna approved these changes Dec 3, 2021

View reviewed changes

vvcephei merged commit 14c2449 into trunk Dec 3, 2021

vvcephei deleted the iqv2-framework branch December 3, 2021 18:53

vvcephei mentioned this pull request Dec 4, 2021

Merge changelog consistency vpapavas/kafka#1

Open

3 tasks

		* This is defined as the metadata of the record that is currently being
		* processed (or was last processed) by the StreamTask that holds the store.

KAFKA-13491: IQv2 framework #11557

KAFKA-13491: IQv2 framework #11557

Conversation

vvcephei commented Nov 30, 2021

Committer Checklist (excluded from commit message)

vvcephei left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

patrickstuedi Dec 1, 2021 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vpapavas Dec 1, 2021 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vvcephei commented Dec 1, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cadonna left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vvcephei commented Dec 3, 2021

vvcephei commented Dec 3, 2021

patrickstuedi Dec 1, 2021 •

edited

vpapavas Dec 1, 2021 •

edited