[BEAM-10212] Add caching state client wrapper #15170

anthonyqzhu · 2021-07-14T20:17:44Z

Introduces a caching state wrapper that stores delegate state client responses in the provided cache.

Unit tests test caching of user and side input state, and test that appends and clears are reflected in the cache.

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Choose reviewer(s) and mention them in a comment (R: @username).
Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

`ValidatesRunner` compliance status (on master branch)

Lang	ULR	Samza	Twister2
Go	---		---
Java
Python	---	---	---
XLang		---	---

Examples testing status on various runners

Lang	ULR	Dataflow	Flink	Samza	Spark	Twister2
Go	---	---	---	---	---	---	---
Java	---		---	---	---	---	---
Python	---	---	---	---	---	---	---
XLang	---	---	---	---	---	---	---

Post-Commit SDK/Transform Integration Tests Status (on master branch)

Go	Java	Python

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website	Whitespace	Typescript
Non-portable
Portable	---			---	---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

anthonyqzhu · 2021-07-14T20:36:14Z

R: @amaliujia

anthonyqzhu · 2021-07-14T20:38:07Z

R: @lukecwik

amaliujia · 2021-07-14T20:58:07Z

cc @tysonjh

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/state/CachingBeamFnStateClient.java

lukecwik · 2021-07-14T20:58:29Z

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/state/CachingBeamFnStateClient.java

+
+    switch (request.getRequestCase()) {
+      case GET:
+        if (ByteString.EMPTY.equals(request.getGet().getContinuationToken())) {


This looks like we are only ever caching the first page and subsequent ones aren't handled.

So should we cache all pages or just the first page?

lukecwik · 2021-07-14T21:12:17Z

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/state/CachingBeamFnStateClient.java

+        if (cachedFirstPage == null) {
+          CompletableFuture<StateResponse> responseFuture = new CompletableFuture<>();
+          beamFnStateClient.handle(
+              StateRequest.newBuilder()


Until we get the rest working, let us treat APPEND as a cache eviction instead. Handling APPEND is more complicated since we want to merge existing data with the new data without needing to have a "get" call and there are a couple of scenarios (e.g., there is no existing data, the last page is still cached, ...).

left append as a regular call to the state client for now

anthonyqzhu · 2021-07-16T17:58:27Z

Run Java PreCommit

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/state/CachingBeamFnStateClient.java

tysonjh · 2021-07-16T21:52:02Z

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/state/CachingBeamFnStateClient.java

+  }
+
+  @AutoValue
+  public abstract static class StateCacheKey {


Add comments please. Note: every class, public method, and tricky bits of code should have comments. There are some exceptions of course, like getters/setters or trivial self explanatory methods.

If you're curious about learning more, you can refer to Google's public style guide for some good practices: https://google.github.io/styleguide/javaguide.html

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/state/CachingBeamFnStateClient.java

anthonyqzhu · 2021-07-19T17:24:37Z

@lukecwik @amaliujia @tysonjh made some fix-up changes, let me know if you have any other suggestions!

amaliujia · 2021-07-19T22:35:20Z

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/state/CachingBeamFnStateClient.java

+        // If data is not cached, add callback to add response to cache on completion.
+        // Otherwise, complete the response with the cached data.
+        if (cachedPage == null) {
+          beamFnStateClient.handle(requestBuilder, response);


Question: how is the GET served when cache miss? When cache hits, you have response.complete, do you need the same for cache miss after you put the state into the cache?

The response.complete is executed by the beamFnStateClient.handle call here, but if it is a cache hit we can complete the response immediately so we do not need to forward the request to the delegate state client

Then can you confirm that after complete, your thenAccept will still be executed? (I am not a async API expert so I am not sure).

amaliujia

Overall LGTM. Left a question for the async logic when cache misses.

@lukecwik @tysonjh do you have any additional comment before this PR is merged.

lukecwik · 2021-07-19T22:48:17Z

Overall LGTM. Left a question for the async logic when cache misses.

@lukecwik @tysonjh do you have any additional comment before this PR is merged.

Yes, I would like to take another look.

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/state/CachingBeamFnStateClient.java

lukecwik · 2021-07-19T23:05:41Z

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/state/CachingBeamFnStateClient.java

+public class CachingBeamFnStateClient implements BeamFnStateClient {
+
+  private final BeamFnStateClient beamFnStateClient;
+  private final LoadingCache<StateKey, Map<StateCacheKey, StateGetResponse>> stateCache;


It doesn't seem like we are using the Loading part of LoadingCache, did you mean to have the CachingBeamFnStateClient perform the loading?

This would generally be a good thing since requests that go for the same state key could resolve down to a single request to the runner instead of multiple.

Leaving this for a future PR

anthonyqzhu · 2021-07-21T17:44:11Z

retest this please

anthonyqzhu · 2021-07-21T18:57:28Z

@lukecwik addressed comments if you want to take another look

amaliujia · 2021-07-21T21:14:17Z

Run Java PreCommit

amaliujia · 2021-07-22T16:47:01Z

Run Java PreCommit

amaliujia · 2021-07-22T19:52:38Z

As Luke is OOO for weeks and comments have been addressed. I will merge this PR to unblock Anthony. If there are still other issues, we can address those in patches.

Failed Java Tests seems have been fixed. I will give it another try.

amaliujia · 2021-07-22T20:57:33Z

Run Java_Examples_Dataflow_Java11 PreCommit

* [BEAM-10212] Add caching state client wrapper.

[BEAM-10212] Add caching state client wrapper

a49606b

lukecwik reviewed Jul 14, 2021

View reviewed changes

anthonyqzhu added 5 commits July 15, 2021 23:24

Add Autovalue Cachekey and address minor comments

520ed93

Change mapping keys and change to cache all pages

e6ab75f

Update unit tests

ff8b04f

Use empty byte string instead of null and apply spotlessJava

ed3ea2a

Switch to LoadingCache

bb74e51

Fix spotbugs check

017d3d3

tysonjh reviewed Jul 16, 2021

View reviewed changes

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/state/CachingBeamFnStateClient.java Outdated Show resolved Hide resolved

anthonyqzhu added 2 commits July 19, 2021 16:37

Add comments

734d6bb

Fix checkstyleMain test

72dd967

anthonyqzhu requested review from amaliujia and lukecwik July 19, 2021 17:20

amaliujia reviewed Jul 19, 2021

View reviewed changes

lukecwik reviewed Jul 19, 2021

View reviewed changes

anthonyqzhu added 3 commits July 21, 2021 17:01

Make caching callback non blocking

c9ec8b6

Invalidate last page after append

32a0881

Simplify tests

cd72ce4

amaliujia closed this Jul 21, 2021

amaliujia reopened this Jul 21, 2021

amaliujia closed this Jul 22, 2021

amaliujia reopened this Jul 22, 2021

amaliujia approved these changes Jul 22, 2021

View reviewed changes

amaliujia merged commit 63257ed into apache:master Jul 22, 2021

ryanthompson591 pushed a commit to ryanthompson591/beam that referenced this pull request Aug 10, 2021

[BEAM-10212] Add caching state client wrapper (apache#15170)

5382734

* [BEAM-10212] Add caching state client wrapper.

calvinleungyk pushed a commit to calvinleungyk/beam that referenced this pull request Sep 22, 2021

[BEAM-10212] Add caching state client wrapper (apache#15170)

5b71b2d

* [BEAM-10212] Add caching state client wrapper.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[BEAM-10212] Add caching state client wrapper #15170

[BEAM-10212] Add caching state client wrapper #15170

anthonyqzhu commented Jul 14, 2021

anthonyqzhu commented Jul 14, 2021

anthonyqzhu commented Jul 14, 2021

amaliujia commented Jul 14, 2021

lukecwik Jul 14, 2021

amaliujia Jul 14, 2021

lukecwik Jul 14, 2021

anthonyqzhu Jul 15, 2021

anthonyqzhu commented Jul 16, 2021

tysonjh Jul 16, 2021

anthonyqzhu commented Jul 19, 2021

amaliujia Jul 19, 2021

anthonyqzhu Jul 19, 2021

amaliujia Jul 19, 2021

amaliujia left a comment

lukecwik commented Jul 19, 2021

lukecwik Jul 19, 2021

anthonyqzhu Jul 20, 2021

anthonyqzhu commented Jul 21, 2021

anthonyqzhu commented Jul 21, 2021

amaliujia commented Jul 21, 2021

amaliujia commented Jul 22, 2021

amaliujia commented Jul 22, 2021

amaliujia commented Jul 22, 2021

[BEAM-10212] Add caching state client wrapper #15170

[BEAM-10212] Add caching state client wrapper #15170

Conversation

anthonyqzhu commented Jul 14, 2021

ValidatesRunner compliance status (on master branch)

Examples testing status on various runners

Post-Commit SDK/Transform Integration Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

GitHub Actions Tests Status (on master branch)

anthonyqzhu commented Jul 14, 2021

anthonyqzhu commented Jul 14, 2021

amaliujia commented Jul 14, 2021

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

anthonyqzhu commented Jul 16, 2021

Choose a reason for hiding this comment

anthonyqzhu commented Jul 19, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

amaliujia left a comment

Choose a reason for hiding this comment

lukecwik commented Jul 19, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

anthonyqzhu commented Jul 21, 2021

anthonyqzhu commented Jul 21, 2021

amaliujia commented Jul 21, 2021

amaliujia commented Jul 22, 2021

amaliujia commented Jul 22, 2021

amaliujia commented Jul 22, 2021

`ValidatesRunner` compliance status (on master branch)