Streaming results for List Keys/Buckets, 2i, and MR #677

alexmoore · 2016-10-19T16:46:21Z

This PR adds support for streaming results in:

List Keys Command + Operation
List Buckets Command + Operation
Secondary Index Commands + Operation
Map Reduce Commands + Operation
Full Bucket Read Command (Riak TS only currently)

This was a very interesting exercise in creating a new results return method from top-to-bottom, while not breaking/changing existing APIs.

Unit tests for your change
Integration tests
Documentation in code
A docs.basho.com PR for public api changes: Not Quite Ready - Adding Java 2i streaming example basho_docs#2270

lukebakken · 2016-10-19T23:10:48Z

src/main/java/com/basho/riak/client/api/commands/ChunkedResponseIterator.java

+    @Override
+    public boolean hasNext()
+    {
+        if(currentIteratorHasNext())


lukebakken · 2016-10-19T23:12:08Z

src/main/java/com/basho/riak/client/api/commands/ImmediateCoreFutureAdapter.java

@@ -0,0 +1,83 @@
+/*


These headers should either be nuked or updated.

jbrisbin · 2016-10-20T13:18:52Z

Looks like a a good first step. IMO it would be good to have as an overall goal to make more use of lambdas, not just as callbacks for asynchronous functions but as a tool in the quiver to reduce garbage. If you follow the code path through a request you see a lot of objects being created to support a single operation. Some of that can't be avoided but there are places I think a lambda-centric approach (versus creating a concrete implementation of some abstract class and instantiating it) would be more efficient.

That said, we're probably not at a level where heap usage, GC, and those other considerations are very high priority.

alexmoore · 2016-10-20T13:30:24Z

Indeed, I was lucky to get away with only 4 new class files, and I think I
did a fairly good job about reuse, but there is always room to improve.

On Thu, Oct 20, 2016 at 9:18 AM, Jon Brisbin [email protected]
wrote:

Looks like a a good first step. IMO it would be good to have as an overall
goal to make more use of lambdas, not just as callbacks for asynchronous
functions but as a tool in the quiver to reduce garbage. If you follow the
code path through a request you see a lot of objects being created to
support a single operation. Some of that can't be avoided but there are
places I think a lambda-centric approach (versus creating a concrete
implementation of some abstract class and instantiating it) would be more
efficient.

That said, we're probably not at a level where heap usage, GC, and those
other considerations are very high priority.

—
You are receiving this because you authored the thread.
Reply to this email directly, view it on GitHub
#677 (comment),
or mute the thread
https://github.com/notifications/unsubscribe-auth/AAAFnomhF4vXxMnhl1wJtVbqoswJi1u7ks5q12o8gaJpZM4KbNBC
.

srgg · 2016-10-20T13:41:52Z

src/main/java/com/basho/riak/client/api/commands/ChunkedResponseIterator.java

+    @Override
+    public synchronized FinalT next()
+    {
+        return createNext.apply(currentIterator.next());


Lack of synchronization on hasNext() whereas next() is synchronized looks looks very suspicious for me.

We may not need the synchronization here, as iterators aren't typically guaranteed to be reentrant. The resulting iterator is single-use only too, so I'll need to make sure I document that somewhere as well.

Originally I did all the lower-level iterator checking/loading in next() instead of hasNext() like it is now. I'll just remove it based on my previous comment.

srgg · 2016-10-20T13:42:56Z

src/main/java/com/basho/riak/client/api/commands/ChunkedResponseIterator.java

+        return continuation != null || possibleChunksRemaining();
+    }
+
+    public BinaryValue getContinuation()


Why don't us get rid of BinaryValue here?

Continuation values (if the command/operation uses them) are always returned as BinaryValue through our interface. The continuation also comes as part of one of the chunked responses, so we need to watch out for it.

srgg · 2016-10-20T13:52:17Z

src/main/java/com/basho/riak/client/api/commands/ChunkedResponseIterator.java

+    private final TransferQueue<ChunkT> chunkQueue;
+    private final Function<CoreT, FinalT> createNext;
+    private final Function<ChunkT, Iterator<CoreT>> getNextIterator;
+    private final Function<ChunkT, BinaryValue> getContinuationFn;


Is there a reason why it may require to expose BinaryValue as a part of getContinuationFn() signature?

srgg · 2016-10-20T14:33:52Z

src/main/java/com/basho/riak/client/api/commands/ChunkedResponseIterator.java

+        catch (InterruptedException e)
+        {
+            Thread.currentThread().interrupt();
+            throw new RuntimeException(e);


Changing an interruption behavior by re-throwing RuntimeException calling interrupt() in the same time is not a common pattern. I have no clear understanding what we-ve got as a result. Theoretically, the most possible behavior is -- the original InterruptedException will be swallowed.

FWIW, I didn't grok this either and just assumed it was the right way 🍤

So what I settled on was to mark the interrupt() flag on the thread and to wrap it in a RuntimeException(), that way the thread should give up control.

If we get an interrupt here it means that the user thread really wanted to interrupt for whatever reason, and we can't block indefinitely while we want for another chunk. In the "Riak is being slow", or "the user did a very bad/slow query" case, blocking indefinitely could deadlock the user thread for longer than wanted.

Also we can't extend hasNext() signature to pass along the InterruptedException, hence the wrap. I need to add a test to see if the user calls cancel() if the future/iterator will pick it up and stop iteration.

@A1 The more I think about the more I feel like there should be enough to throw teh only Runtime exception. We can't throw Interrupted due to the contract, therefore nobody will expect that any of it's methods might be blocked and as result nobody will expect that thread may be marked as interrupted. If you approve, I will introduce corresponding changes as a separate commit.

Reading more opinions about this... I guess we could add a flag to note that it was interrupted, and keep going. If we get to the end (hasNext() returns false) we could do something at that point, but not sure what since the iterator interface won't allow us to interrupt.

My initial thought was exactly the same: set the flag and move forward without raising any exception. If this call was made from the whatever execution thread the programmer should take care about the proper thread interuption and at least we do not make it worse.

Please excuse me for my hesitation.

lukebakken · 2016-10-20T16:18:20Z

This all seems fine to a non-Java dev. While I know why the queue and .poll() is used, it seems hacky. Feeble searching for information about netty brought up this SO post which talks about methods to execute callbacks on threads that don't block the IO thread. Maybe something to look into for the future?

srgg · 2016-10-24T13:45:23Z

src/test/java/com/basho/riak/client/api/commands/indexes/itest/ITestIndexBase.java

+
+import java.util.concurrent.ExecutionException;
+
+public abstract class ITestIndexBase extends ITestBase


We already have ITestAutoCleanupBase that do mostly the same

Mostly, but ITestAutoCleanupBase cleans up after every test, while this one cleans up after each suite/class of tests. With some of the tests doing 1K/10K entries it speeds things up a bit to use common data across all a class's tests.

I wanted to also put an abstract setupData() method in here but I couldn't get the RiakCluster/RiakClient dependency and it's initialization correct with all the static context going on.

…akCommand

…signature

… contract

Simplification streamable commands

alexmoore · 2016-11-15T16:21:30Z

@srgg FullBucketRead tests pass locally, +1 to 51bff20. Going to review the rest of it now to see if anything jumps out.

…lPointerException.

srgg · 2016-11-15T19:24:37Z

src/main/java/com/basho/riak/client/api/commands/indexes/SecondaryIndexQuery.java

        public final List<E> getEntries()
        {
+            if(isStreamable())
+            {
+                return new ArrayList<>(0);


What about throwing either IllegalState or UnsupportedOperation?

if not, it is better to return Collections.EMPTY_LIST

…result as a non-streaming one

alexmoore

+1 to any changes I didn't do. Also added some misc public api javadocs and did some whitespace cleanup.

srgg · 2016-11-16T03:43:50Z

+1 to any changes I didn't do.

lukebakken · 2016-11-16T04:58:52Z

Looks like more code was deleted than written, always a good sign. Tests pass with Riak 2.2.0, 2.1.4, and 2.0.7 (except one YZ test there). 👍

alexmoore added 14 commits October 5, 2016 11:00

Add streaming api to operations/futures

e1480aa

Add Streaming List Buckets

af3d5ba

Add streaming list keys support

5c56d9f

Add streaming 2i operations/results

78c9076

Use TransferQueues instead of BlockingQueues

590bcde

Add Streaming MapReduce Operations

5b87e24

Forgot a file

69aed60

Command level streaming for ListKeys and ListBuckets

c7c40dd

Cleanup & name things betterer

ca0aaf4

Start into Streaming 2i special commands.

ca29f1a

Stub out rest of streaming 2i commands

e6de1e0

Add Javadocs, remove need to check for nulls from streaming results

35e9c4f

Streaming 2i Command Integration Tests

82fb339

Implement Streaming MapReduce Results + Test

9ac71b6

alexmoore assigned jbrisbin, lukebakken and srgg Oct 19, 2016

lukebakken reviewed Oct 19, 2016

View reviewed changes

Fix spacing in control structures.

d19cd3e

srgg reviewed Oct 20, 2016

View reviewed changes

Fix headers for newest/recently changed files.

fe24ba1

srgg reviewed Oct 20, 2016

View reviewed changes

Don't need synchronization on next(), removing unused import

3edac1f

srgg reviewed Oct 20, 2016

View reviewed changes

srgg reviewed Oct 24, 2016

View reviewed changes

srgg and others added 16 commits November 7, 2016 16:49

Simplification of a bunch of RiakCommand descendants

9630a11

Simplification for the case when no needs to convert results

63efe16

Generalize streamable responses

52795d4

Get rid of unsafe default response and infor conversions in GenericRi…

3e5c7b0

…akCommand

Get rid of unused GenericRiakCommand constructor

d97183a

Request argument was added to the GenericRiakCommand.convertResponse …

76a6ee4

…signature

Generalize StreamableRiakCommand.executeAsyncStreaming()

7cdf043

Remove if / subtype casting from Command execution

45f55c4

Make PBStreamingFutureOperation to be a part of StreamableRiakCommand…

12c8944

… contract

Refactor streaming responses to re-use ConvertibleIterator

944fb90

Simplify ChunkedResponseIterator initial loading

67e56cb

Generalize StreamableResponse

22655d4

fix iteration through chunked response

3afa145

Merge pull request #684 from basho/simplification-streamable-commands

a07833f

Simplification streamable commands

Adding assert around chunk queue insertion

e313956

Adding integration tests for streamable full bucket read.

51bff20

alexmoore added 2 commits November 15, 2016 14:21

Fix bug where calling 2i getEntries() while streaming would throw Nul…

002ba0e

…lPointerException.

Pedantic whitespace formatting.

a13798b

srgg reviewed Nov 15, 2016

View reviewed changes

alexmoore added 4 commits November 15, 2016 14:36

Merge branch 'develop' into streaming-api-2

c1b575a

Switch 2i getEntries() to throw IllegalStateException.

ac3d7e2

Adding javadocs to new public API pieces

78c43f1

Changing isStreamable to isStreaming, as we cannot treat a streaming …

a797ba6

…result as a non-streaming one

alexmoore commented Nov 15, 2016

View reviewed changes

alexmoore merged commit e385fc3 into develop Nov 16, 2016

alexmoore deleted the streaming-api-2 branch November 16, 2016 15:25

antn unassigned lukebakken Jun 15, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Streaming results for List Keys/Buckets, 2i, and MR #677

Streaming results for List Keys/Buckets, 2i, and MR #677

alexmoore commented Oct 19, 2016 •

edited

Loading

lukebakken Oct 19, 2016

lukebakken Oct 19, 2016

alexmoore Oct 20, 2016

jbrisbin commented Oct 20, 2016

alexmoore commented Oct 20, 2016

srgg Oct 20, 2016

alexmoore Oct 20, 2016

alexmoore Oct 20, 2016

srgg Oct 20, 2016

alexmoore Oct 20, 2016

srgg Oct 20, 2016

alexmoore Oct 20, 2016

srgg Oct 20, 2016 •

edited

Loading

lukebakken Oct 20, 2016

alexmoore Oct 20, 2016

srgg Oct 26, 2016 •

edited

Loading

alexmoore Oct 26, 2016 •

edited

Loading

srgg Oct 27, 2016

lukebakken commented Oct 20, 2016

srgg Oct 24, 2016 •

edited

Loading

alexmoore Oct 24, 2016

alexmoore commented Nov 15, 2016

srgg Nov 15, 2016

srgg Nov 15, 2016

alexmoore left a comment

srgg commented Nov 16, 2016

lukebakken commented Nov 16, 2016


		import java.util.concurrent.ExecutionException;

		public abstract class ITestIndexBase extends ITestBase

Streaming results for List Keys/Buckets, 2i, and MR #677

Streaming results for List Keys/Buckets, 2i, and MR #677

Conversation

alexmoore commented Oct 19, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jbrisbin commented Oct 20, 2016

alexmoore commented Oct 20, 2016

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

srgg Oct 20, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

srgg Oct 26, 2016 • edited Loading

Choose a reason for hiding this comment

alexmoore Oct 26, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lukebakken commented Oct 20, 2016

srgg Oct 24, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alexmoore commented Nov 15, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alexmoore left a comment

Choose a reason for hiding this comment

srgg commented Nov 16, 2016

lukebakken commented Nov 16, 2016

alexmoore commented Oct 19, 2016 •

edited

Loading

srgg Oct 20, 2016 •

edited

Loading

srgg Oct 26, 2016 •

edited

Loading

alexmoore Oct 26, 2016 •

edited

Loading

srgg Oct 24, 2016 •

edited

Loading