Rs3 Window Operations #443

hachikuji · 2025-04-03T18:58:09Z

Adds support for window tables. Basically a big bunch of banal boilerplate to send window timestamps as needed in the GRPC protocol.

ableegoldman

Mostly minor comments, from a high level this all looks good

kafka-client/src/main/java/dev/responsive/kafka/internal/db/rs3/RS3TableFactory.java

kafka-client/src/main/java/dev/responsive/kafka/internal/db/rs3/RS3WindowFlushManager.java

ableegoldman · 2025-04-03T22:07:08Z

kafka-client/src/main/java/dev/responsive/kafka/internal/db/rs3/RS3WindowFlushManager.java

+      final long batchOffset,
+      final Integer failedTablePartition
+  ) {
+    return String.format("");


I forget how exactly this gets used but we probably want to log at least the batch offset and failedTablePartition, probably also the current streamTime. And definitely the table name if it's not already included in the error message (maybe compare other implementations to see what's included in this info and what isn't)

ableegoldman · 2025-04-03T22:12:28Z

kafka-client/src/main/java/dev/responsive/kafka/internal/db/rs3/RS3WindowFlushManager.java

+
+  @Override
+  public void writeAdded(final WindowedKey key) {
+    streamTime = Long.max(streamTime, key.windowStartMs);


I know it's not actually being used anywhere at the moment so it doesn't matter too much, but to avoid confusion in the future we might want to name this more specifically to indicate that it isn't the persisted streamTime but rather the stream-time of the current pending batch (eg batchStreamTime or pendingStreamTime or something better)

OTOH just thinking about how we'll want to use the streamTime in the future, maybe we actually do want the persisted streamTime only, to ensure it monotonically increases? Or will slateDB handle that for us automatically 🤔

kafka-client/src/main/java/dev/responsive/kafka/internal/db/rs3/RS3WindowTable.java

kafka-client/src/main/java/dev/responsive/kafka/internal/db/rs3/RS3WindowedKeySerde.java

kafka-client/src/main/java/dev/responsive/kafka/internal/db/rs3/RS3Writer.java

ableegoldman · 2025-04-03T22:23:34Z

kafka-client/src/main/java/dev/responsive/kafka/internal/stores/RemoteWindowOperations.java

+        final var scopeBuilder = responsiveMetrics.storeLevelMetricScopeBuilder(
+            Utils.extractThreadIdFromThreadName(Thread.currentThread().getName()),
+            changelog,
+            params.name().tableName()


I know we use #tableName for the existing caller of this method but I think it should actually be #kafkaName (here and in PartitionedOperations)

We seem to use the kafka store name rather than the table name for these metrics elsewhere

ableegoldman · 2025-04-03T22:28:38Z

kafka-client/src/main/java/dev/responsive/kafka/internal/stores/RemoteWindowOperations.java

+      throw new UnsupportedOperationException("Duplicate retention is not yet supported in RS3");
+    }
+
+    // TODO: Pass through retention period once we have support for TTL


FYI I'm actually doing this as part of the CreateStore implementation

It's just setting the default ttl for the store (so we still need the STREAM_TIME clock type for full window store ttl support) but we can do the same thing here. So I can handle this as part of integrating the CreateStore API with the RS3 window stores after both PRs are merged

hachikuji · 2025-04-03T23:32:08Z

As I was discussing the patch with @ableegoldman a bit, I realized there is a fork in the road that I probably haven't given enough consideration to. I've taken the approach here of making the RS3 server agnostic to window store semantics. The window start time is baked directly into the key that gets shipped to RS3 and the server knows nothing about it. An alternative is to involve the server directly. We could pass through the window start time explicitly in the WriteWalSegment API and we could leave it up to the RS3 server to decide how to encode it in the data. We could then modify the range API to additionally pass through the window start time that is being queried. This is a bit more future proof since the data format would be encapsulated in RS3. But the larger point is allowing RS3 to understand window semantics, which may be useful when trying to optimize performance or just for trying to debug/understand usage. For example, it would allow us to create time-based indexes without having to change client code.

…3/RS3WindowFlushManager.java Co-authored-by: A. Sophie Blee-Goldman <[email protected]>

…3/RS3WindowTable.java Co-authored-by: A. Sophie Blee-Goldman <[email protected]>

ableegoldman · 2025-04-15T21:58:34Z

kafka-client/src/main/java/dev/responsive/kafka/internal/db/rs3/RS3TableFactory.java

+    );
+  }
+
+  private UUID lookupStoreId(


fyi i'm just about to merge the rs3 PR to convert store ids from UUID to string, haven't done the responsive-pub one yet but it probably makes sense to just drop the UUID and use string for the window store now so you don't have to rebase this

kafka-client/src/main/java/dev/responsive/kafka/internal/stores/ResponsiveWindowStore.java

ableegoldman

LGTM!

ableegoldman · 2025-04-25T23:02:04Z

kafka-client/src/main/java/dev/responsive/kafka/internal/db/rs3/RS3TableFactory.java

+    final UUID storeId = createdStores.computeIfAbsent(storeName, n -> createStore(
+        storeName,
+        CreateStoreTypes.StoreType.WINDOW,
+        Optional.of(ClockType.STREAM_TIME),


just a heads up we still throw a ConfigErr on the server side if stream time is used so we'll need to implement that as a followup

kafka-client/src/main/java/dev/responsive/kafka/internal/db/rs3/RS3WindowFlushManager.java

hachikuji added 4 commits April 3, 2025 11:52

Initial implementation of RS3 window table

832bb61

Merge remote-tracking branch 'origin/main' into rs3-window-operations

e30d0ab

Remove unneeded comment

d50d841

Test cases for serde

db3e73d

ableegoldman reviewed Apr 3, 2025

View reviewed changes

Update kafka-client/src/main/java/dev/responsive/kafka/internal/db/rs…

68e1718

…3/RS3WindowFlushManager.java Co-authored-by: A. Sophie Blee-Goldman <[email protected]>

hachikuji mentioned this pull request Apr 4, 2025

Rename fetchOffset to lastWrittenOffset #444

Merged

hachikuji and others added 14 commits April 3, 2025 19:26

use more final variables

8d47b4d

review comments

2770de1

Remove TODO

0cb4b5e

Update kafka-client/src/main/java/dev/responsive/kafka/internal/db/rs…

d47f03a

…3/RS3WindowTable.java Co-authored-by: A. Sophie Blee-Goldman <[email protected]>

reduce visibility on RS3Writer

4452308

Add more info to fail message

a1e7916

Add basic writer test for window table

2a6e247

Merge remote-tracking branch 'origin/main' into rs3-window-operations

8b5d2f5

Fix breakage from merge

7c8fdd1

Add range implementations for window table

e46e96f

Update rs3 proto

b385f5e

Fix failing tests

03bdf57

Merge remote-tracking branch 'origin/main' into rs3-window-operations

d210126

Reset opentelemetry to main branch

3e7ab01

ableegoldman reviewed Apr 15, 2025

View reviewed changes

hachikuji added 5 commits April 23, 2025 11:15

Merge remote-tracking branch 'origin/main' into rs3-window-operations

2a1e26f

Fix breakage wip

c158a38

Closer to compiling

b07c812

compiling again

a9d7615

Fix checkstyle

173b798

hachikuji commented Apr 24, 2025

View reviewed changes

kafka-client/src/main/java/dev/responsive/kafka/internal/stores/ResponsiveWindowStore.java Outdated Show resolved Hide resolved

Pass through default ttl/use stream time clock

cfa3e71

hachikuji added 7 commits April 25, 2025 09:58

Fix failing end to end tests

e188cd8

Implement windowed end-to-end test

873b644

Merge remote-tracking branch 'origin/main' into rs3-window-operations

0728022

fix merge breakage

e2ae90c

Fix remaining broken tests

9a8045e

Fix window value iteration with test case

68e9fa8

Merge remote-tracking branch 'origin/main' into rs3-window-operations

f80d98d

ableegoldman approved these changes Apr 25, 2025

View reviewed changes

hachikuji added 3 commits April 25, 2025 16:31

Get rid of TODO

b93a539

Don't commit jwik database

4c5ee03

Use wall clock time for now

db8fe92

hachikuji merged commit 192eaea into main Apr 26, 2025
1 check passed

hachikuji deleted the rs3-window-operations branch April 26, 2025 02:17

Rs3 Window Operations #443

Rs3 Window Operations #443

Uh oh!

Conversation

hachikuji commented Apr 3, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ableegoldman left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

ableegoldman Apr 3, 2025

Choose a reason for hiding this comment

Uh oh!

ableegoldman Apr 3, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

ableegoldman Apr 3, 2025

Choose a reason for hiding this comment

Uh oh!

ableegoldman Apr 3, 2025

Choose a reason for hiding this comment

Uh oh!

hachikuji commented Apr 3, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ableegoldman Apr 15, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

ableegoldman left a comment

Choose a reason for hiding this comment

Uh oh!

ableegoldman Apr 25, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

hachikuji commented Apr 3, 2025 •

edited

Loading

hachikuji commented Apr 3, 2025 •

edited

Loading