You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "bbejeck (via GitHub)" <gi...@apache.org> on 2026/03/08 01:16:37 UTC
[PR] KAFKA-20277: Add session store with headers infra and support for IQv2 [kafka]
bbejeck opened a new pull request, #21670:
URL: https://github.com/apache/kafka/pull/21670
This PR is NOT about implementing IQv2s for header-store, but provides
IQv2s for headers store through the session state stores. So the
results do not contain the headers even though the headers are preserved
in the headers state store.
Part of KIP-1271.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20277: Add session store with headers support for IQv2 [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on PR #21670:
URL: https://github.com/apache/kafka/pull/21670#issuecomment-4041909218
Merged #21670 into trunk
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20277: Add session store with headers support for IQv2 [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck merged PR #21670:
URL: https://github.com/apache/kafka/pull/21670
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20277: Add session store with headers support for IQv2 [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21670:
URL: https://github.com/apache/kafka/pull/21670#discussion_r2915173211
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreWithHeaders.java:
##########
@@ -59,6 +68,84 @@ public void put(final Windowed<K> sessionKey, final AggregationWithHeaders<AGG>
}
+ @Override
+ public KeyValueIterator<Windowed<K>, AggregationWithHeaders<AGG>> findSessions(final K key,
+ final Instant earliestSessionEndTime,
+ final Instant latestSessionStartTime) {
+ return super.findSessions(key, earliestSessionEndTime, latestSessionStartTime);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public <R> QueryResult<R> query(final Query<R> query,
+ final PositionBound positionBound,
+ final QueryConfig config) {
+ final long start = time.nanoseconds();
+ final QueryResult<R> result;
+
+ if (query instanceof WindowRangeQuery) {
+ final WindowRangeQuery<?, ?> windowRangeQuery = (WindowRangeQuery<?, ?>) query;
+ if (windowRangeQuery.getKey().isPresent()) {
+ result = runRangeQueryWithHeadersUnwrap(query, positionBound, config);
Review Comment:
```suggestion
result = runRangeQuery(query, positionBound, config);
```
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/PositionRestartIntegrationTest.java:
##########
@@ -248,6 +248,7 @@ public boolean isWindowed() {
public boolean isSession() {
return false;
}
+
Review Comment:
Can we revert this to get rid of the whole file in the PR?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreWithHeaders.java:
##########
@@ -59,6 +68,84 @@ public void put(final Windowed<K> sessionKey, final AggregationWithHeaders<AGG>
}
+ @Override
+ public KeyValueIterator<Windowed<K>, AggregationWithHeaders<AGG>> findSessions(final K key,
+ final Instant earliestSessionEndTime,
+ final Instant latestSessionStartTime) {
+ return super.findSessions(key, earliestSessionEndTime, latestSessionStartTime);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public <R> QueryResult<R> query(final Query<R> query,
+ final PositionBound positionBound,
+ final QueryConfig config) {
+ final long start = time.nanoseconds();
+ final QueryResult<R> result;
+
+ if (query instanceof WindowRangeQuery) {
+ final WindowRangeQuery<?, ?> windowRangeQuery = (WindowRangeQuery<?, ?>) query;
+ if (windowRangeQuery.getKey().isPresent()) {
+ result = runRangeQueryWithHeadersUnwrap(query, positionBound, config);
+ } else {
+ result = QueryResult.forFailure(
+ FailureReason.UNKNOWN_QUERY_TYPE,
+ "This store (" + getClass() + ") doesn't know how to"
+ + " execute the given query (" + query + ") because"
+ + " SessionStores only support WindowRangeQuery.withKey."
+ + " Contact the store maintainer if you need support"
+ + " for a new query type."
+ );
+ }
+ if (config.isCollectExecutionInfo()) {
+ result.addExecutionInfo(
+ "Handled in " + getClass() + " with serdes "
+ + serdes + " in " + (time.nanoseconds() - start) + "ns");
+ }
+ } else {
+ result = wrapped().query(query, positionBound, config);
+ if (config.isCollectExecutionInfo()) {
+ result.addExecutionInfo(
+ "Handled in " + getClass() + " in " + (time.nanoseconds() - start) + "ns");
+ }
+ }
+ return result;
+ }
+
+ @SuppressWarnings("unchecked")
+ private <R> QueryResult<R> runRangeQueryWithHeadersUnwrap(final Query<R> query,
+ final PositionBound positionBound,
+ final QueryConfig config) {
+ final WindowRangeQuery<K, ?> typedQuery = (WindowRangeQuery<K, ?>) query;
+ final WindowRangeQuery<Bytes, byte[]> rawKeyQuery =
+ WindowRangeQuery.withKey(
+ Bytes.wrap(serdes.rawKey(typedQuery.getKey().get(), new RecordHeaders()))
+ );
+ final QueryResult<KeyValueIterator<Windowed<Bytes>, byte[]>> rawResult =
+ wrapped().query(rawKeyQuery, positionBound, config);
+ if (rawResult.isSuccess()) {
+ final MeteredWindowedKeyValueIterator<K, ?> typedResult =
Review Comment:
`?` ?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreWithHeaders.java:
##########
@@ -59,6 +68,84 @@ public void put(final Windowed<K> sessionKey, final AggregationWithHeaders<AGG>
}
+ @Override
+ public KeyValueIterator<Windowed<K>, AggregationWithHeaders<AGG>> findSessions(final K key,
+ final Instant earliestSessionEndTime,
+ final Instant latestSessionStartTime) {
+ return super.findSessions(key, earliestSessionEndTime, latestSessionStartTime);
Review Comment:
Why are we overriding this method if we only call `super` anyway?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreWithHeaders.java:
##########
@@ -59,6 +68,84 @@ public void put(final Windowed<K> sessionKey, final AggregationWithHeaders<AGG>
}
+ @Override
+ public KeyValueIterator<Windowed<K>, AggregationWithHeaders<AGG>> findSessions(final K key,
+ final Instant earliestSessionEndTime,
+ final Instant latestSessionStartTime) {
+ return super.findSessions(key, earliestSessionEndTime, latestSessionStartTime);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public <R> QueryResult<R> query(final Query<R> query,
+ final PositionBound positionBound,
+ final QueryConfig config) {
+ final long start = time.nanoseconds();
+ final QueryResult<R> result;
+
+ if (query instanceof WindowRangeQuery) {
+ final WindowRangeQuery<?, ?> windowRangeQuery = (WindowRangeQuery<?, ?>) query;
Review Comment:
Why are we using `<?, ?>` instead of `<K, V>` like we do in `MeteredSessionStore`.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreWithHeaders.java:
##########
@@ -59,6 +68,84 @@ public void put(final Windowed<K> sessionKey, final AggregationWithHeaders<AGG>
}
+ @Override
+ public KeyValueIterator<Windowed<K>, AggregationWithHeaders<AGG>> findSessions(final K key,
+ final Instant earliestSessionEndTime,
+ final Instant latestSessionStartTime) {
+ return super.findSessions(key, earliestSessionEndTime, latestSessionStartTime);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public <R> QueryResult<R> query(final Query<R> query,
+ final PositionBound positionBound,
+ final QueryConfig config) {
+ final long start = time.nanoseconds();
+ final QueryResult<R> result;
+
+ if (query instanceof WindowRangeQuery) {
+ final WindowRangeQuery<?, ?> windowRangeQuery = (WindowRangeQuery<?, ?>) query;
+ if (windowRangeQuery.getKey().isPresent()) {
+ result = runRangeQueryWithHeadersUnwrap(query, positionBound, config);
Review Comment:
Let's keep the name simple -- it's a `private` method only, so it seems unnecessary to have a too complex name.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredSessionStoreWithHeaders.java:
##########
@@ -59,6 +68,84 @@ public void put(final Windowed<K> sessionKey, final AggregationWithHeaders<AGG>
}
+ @Override
+ public KeyValueIterator<Windowed<K>, AggregationWithHeaders<AGG>> findSessions(final K key,
+ final Instant earliestSessionEndTime,
+ final Instant latestSessionStartTime) {
+ return super.findSessions(key, earliestSessionEndTime, latestSessionStartTime);
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public <R> QueryResult<R> query(final Query<R> query,
+ final PositionBound positionBound,
+ final QueryConfig config) {
+ final long start = time.nanoseconds();
+ final QueryResult<R> result;
+
+ if (query instanceof WindowRangeQuery) {
+ final WindowRangeQuery<?, ?> windowRangeQuery = (WindowRangeQuery<?, ?>) query;
+ if (windowRangeQuery.getKey().isPresent()) {
+ result = runRangeQueryWithHeadersUnwrap(query, positionBound, config);
+ } else {
+ result = QueryResult.forFailure(
+ FailureReason.UNKNOWN_QUERY_TYPE,
+ "This store (" + getClass() + ") doesn't know how to"
+ + " execute the given query (" + query + ") because"
+ + " SessionStores only support WindowRangeQuery.withKey."
+ + " Contact the store maintainer if you need support"
+ + " for a new query type."
+ );
+ }
+ if (config.isCollectExecutionInfo()) {
+ result.addExecutionInfo(
+ "Handled in " + getClass() + " with serdes "
+ + serdes + " in " + (time.nanoseconds() - start) + "ns");
+ }
+ } else {
+ result = wrapped().query(query, positionBound, config);
+ if (config.isCollectExecutionInfo()) {
+ result.addExecutionInfo(
+ "Handled in " + getClass() + " in " + (time.nanoseconds() - start) + "ns");
+ }
+ }
+ return result;
+ }
+
+ @SuppressWarnings("unchecked")
+ private <R> QueryResult<R> runRangeQueryWithHeadersUnwrap(final Query<R> query,
+ final PositionBound positionBound,
+ final QueryConfig config) {
+ final WindowRangeQuery<K, ?> typedQuery = (WindowRangeQuery<K, ?>) query;
Review Comment:
As above. Why `?` ?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org