KAFKA-5363: KIP-167 implementing bulk load, restoration event notification #3325

bbejeck · 2017-06-13T20:41:09Z

No description provided.

bbejeck · 2017-06-13T20:42:43Z

ping @mjsax @guozhangwang @enothereska @dguy for initial review

This still need more test coverage, but I wanted to submit the PR to move the discussion forward.

EDIT: The batch restoration benchmark will undergo some refactoring as well.

EDIT round 2: Batch restoration benchmark will be in a follow in PR

asfbot · 2017-06-13T20:56:03Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/5236/
Test FAILed (JDK 8 and Scala 2.12).

bbejeck · 2017-06-13T20:57:40Z

Fixing checkstyle errors

asfbot · 2017-06-13T21:43:20Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/5254/
Test FAILed (JDK 7 and Scala 2.11).

asfbot · 2017-06-13T21:45:06Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/5239/
Test FAILed (JDK 8 and Scala 2.12).

bbejeck · 2017-06-13T21:46:00Z

addressing errors

asfbot · 2017-06-13T22:57:39Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/5258/
Test PASSed (JDK 7 and Scala 2.11).

asfbot · 2017-06-13T23:07:01Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/5242/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-06-14T14:08:18Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/5298/
Test FAILed (JDK 7 and Scala 2.11).

bbejeck · 2017-06-14T14:17:19Z

Error seems unrelated to this PR Execution failed for task ':core:compileScala'. Can't reproduce locally either.

asfbot · 2017-06-14T15:13:26Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/5282/
Test PASSed (JDK 8 and Scala 2.12).

bbejeck · 2017-06-14T15:16:21Z

retest this please.

enothereska · 2017-06-14T15:55:50Z

I think the restore callbacks etc look good. My main question will be around the bulk loading part, i.e., if we go with batch loading, will we still be able to look into KAFKA-4868 or are they mutually exclusive? Thanks @bbejeck .

asfbot · 2017-06-14T16:09:40Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/5285/
Test FAILed (JDK 8 and Scala 2.12).

bbejeck · 2017-06-14T16:20:25Z

@enothereska current plan is to have KAFKA-4868 be part of the bulk/batch loading in this PR

asfbot · 2017-06-14T16:47:23Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/5301/
Test FAILed (JDK 7 and Scala 2.11).

bbejeck · 2017-06-14T18:02:37Z

retest this please.

asfbot · 2017-06-14T19:22:03Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/5317/
Test PASSed (JDK 7 and Scala 2.11).

asfbot · 2017-06-14T19:22:06Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/5301/
Test FAILed (JDK 8 and Scala 2.12).

asfgit · 2017-06-15T21:39:42Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/5369/
Test PASSed (JDK 7 and Scala 2.11).

asfgit · 2017-06-15T22:13:46Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/5354/
Test PASSed (JDK 8 and Scala 2.12).

bbejeck · 2017-06-20T01:31:15Z

@enothereska - updated code to handle KAFKA-4868

EDIT: Still owe tests on this PR

asfgit · 2017-06-20T02:32:09Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/5476/
Test PASSed (JDK 8 and Scala 2.12).

asfgit · 2017-06-20T02:32:12Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/5491/
Test PASSed (JDK 7 and Scala 2.11).

asfgit · 2017-06-29T20:41:37Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/5811/
Test PASSed (JDK 7 and Scala 2.11).

asfgit · 2017-06-29T20:47:10Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/5797/
Test PASSed (JDK 8 and Scala 2.12).

bbejeck · 2017-06-30T11:41:16Z

Still owe tests

mjsax

Some more comments...

mjsax · 2017-07-26T15:27:05Z

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

+ * across all {@link org.apache.kafka.streams.processor.internals.StreamThread} instances.
+ *
+ * Users desiring stateful operations will need to provide synchronization internally in
+ * the StateRestorerListener implementation.


Nit {@code StateRestorerListener}

mjsax · 2017-07-26T15:33:46Z

...ams/src/main/java/org/apache/kafka/streams/processor/internals/CompositeRestoreListener.java

+    }
+
+
+    private static final class NoOpStateRestoreListener implements StateRestoreListener {


Can't this extend AbstractNotifyingRestoreCallback to save all the boiler plate from below?

Will still have one no-op method, but I guess it's worth it as it does reduce the boilerplate some.

mjsax · 2017-07-26T16:05:20Z

...c/main/java/org/apache/kafka/streams/processor/AbstractNotifyingBatchingRestoreCallback.java

+     * or {@link StateRestoreCallback} instead for single action restores.
+     */
+    @Override
+    public void restore(byte[] key, byte[] value) {


Can we introduce a global "one parameter per line" code style? I think it would help to make diffs cleaner. We can do this incrementally. If yes, please do for all newly introduced code of this PR.

Also, should be add final all over the place?

Sure thing, can you add to the Steams guidelines?

mjsax · 2017-07-26T16:07:06Z

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

+    /**
+     * Method called at the very beginning of {@link StateStore} restoration.
+     *
+     * @param topicPartition the TopicPartition containing the values to restore.


nit: parameter descriptions are no sentences, thus no . at the end (on many other places, too). If we say they are sentences, they it should start with upper case [T]he TopicPartition

mjsax · 2017-07-26T16:13:55Z

...ams/src/main/java/org/apache/kafka/streams/processor/internals/CompositeRestoreListener.java

+        }
+    }
+
+    private static final class NoOpStateRestoreCallback implements StateRestoreCallback {


Actually this class won't be used anymore, so removed.

mjsax · 2017-07-26T16:24:35Z

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

@@ -135,6 +140,10 @@ public void openDB(ProcessorContext context) {
        // (this could be a bug in the RocksDB code and their devs have been contacted).
        options.setIncreaseParallelism(Math.max(Runtime.getRuntime().availableProcessors(), 2));

+        if (prepareForBulkload) {
+            options.prepareForBulkLoad();


Can you elaborate?

mjsax · 2017-07-26T16:28:29Z

streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java

@@ -254,6 +255,19 @@ public void testCannotStartTwice() throws Exception {
        }
    }

+    @Test(expected = IllegalStateException.class)


We should not expect here and use fail within try-catch

mjsax · 2017-07-26T16:35:25Z

streams/src/test/java/org/apache/kafka/test/MockProcessorContext.java

@@ -305,4 +327,24 @@ public ProcessorNode currentNode() {
    public void close() {
        metrics.close();
    }
+
+    private static final class NoOpRestoreListener implements StateRestoreListener {


bbejeck · 2017-07-26T21:06:23Z

rebasing now

…es, adding state restore listener KAFKA-5363: updates per comments on KIP

…start and update end, updated RocksDB to be closed and opened during restores. Added adapter class for abstracting the callback implementations.

…and StateRestoreCallback

…unctions

bbejeck · 2017-07-26T21:24:04Z

@guozhangwang @mjsax
rebased and comments addressed

asfgit · 2017-07-26T22:06:58Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/6369/
Test PASSed (JDK 7 and Scala 2.11).

asfgit · 2017-07-26T22:23:59Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/6372/
Test PASSed (JDK 7 and Scala 2.11).

asfgit · 2017-07-26T22:29:56Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/6354/
Test PASSed (JDK 8 and Scala 2.12).

asfgit · 2017-07-26T22:43:57Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/6357/
Test PASSed (JDK 8 and Scala 2.12).

bbejeck · 2017-07-27T20:02:09Z

ping @guozhangwang for final review and merge

guozhangwang

@bbejeck Some more minor comments about code styles.

guozhangwang · 2017-07-27T21:26:05Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StateRestorer.java

@@ -51,14 +57,30 @@ long checkpoint() {
        return checkpoint == null ? NO_CHECKPOINT : checkpoint;
    }

-    void restore(final byte[] key, final byte[] value) {
-        stateRestoreCallback.restore(key, value);
+    void notifyRestoreStarted(long startingOffset, long endingOffset) {


nit: rename this function to restoreStarted to be consistent with other names. Such will help other code readers to understand these functions are for the same code granularity and semantics.

guozhangwang · 2017-07-27T21:28:46Z

...ams/src/main/java/org/apache/kafka/streams/processor/internals/CompositeRestoreListener.java

+        if (stateRestoreCallback instanceof StateRestoreListener) {
+            storeRestoreListener = (StateRestoreListener) stateRestoreCallback;
+        } else {
+            storeRestoreListener = new NoOpStateRestoreListener();


Is this comment missed somehow? I think line 42 above could be storeRestoreListener = NO_OP_STATE_RESTORE_LISTENER.

Ack, must have overlooked

guozhangwang · 2017-07-27T21:32:21Z

streams/src/test/java/org/apache/kafka/test/MockStateRestoreListener.java

+
+public class MockStateRestoreListener extends AbstractNotifyingRestoreCallback {
+
+    //Verifies store name called for each state


nit: space after // and we do not need capitalize the in-function comments.

guozhangwang · 2017-07-27T21:33:28Z

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

@@ -198,6 +220,11 @@ private RocksDB openDB(File dir, Options options, int ttl) throws IOException {
        }
    }

+    //Visible for testing


Ditto for in-function and simple top function comments.

guozhangwang · 2017-07-27T21:34:51Z

...ams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java

+    private final MockConsumer<byte[], byte[]> consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
+    private final StateRestoreListener stateRestoreListener = new MockStateRestoreListener();
+    private final StoreChangelogReader
+        changelogReader =


nit: new lines are generally not recommended to break object type declaration with object name. For this specific line I think we can still make them in one line.

Ack, need to adjust Intellij settings

guozhangwang · 2017-07-27T21:35:38Z

...ams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java

@@ -56,7 +71,8 @@ public void shouldThrowStreamsExceptionWhenTimeoutExceptionThrown() throws Excep
                throw new TimeoutException("KABOOM!");
            }
        };
-        final StoreChangelogReader changelogReader = new StoreChangelogReader(consumer, new MockTime(), 0);
+        final StoreChangelogReader changelogReader = new StoreChangelogReader(consumer, new
+            MockTime(), 0, stateRestoreListener);


Ditto: newline after keywords are generally not recommended.

Ack, same as above

guozhangwang · 2017-07-27T21:37:04Z

...va/org/apache/kafka/streams/processor/internals/WrappedBatchingStateRestoreCallbackTest.java

+    private final byte[] value = "value".getBytes(Charset.forName("UTF-8"));
+    private final Collection<KeyValue<byte[], byte[]>> records = Collections.singletonList(KeyValue.pair(key, value));
+    private final BatchingStateRestoreCallback wrappedBatchingStateRestoreCallback = new
+        WrappedBatchingStateRestoreCallback(mockRestoreCallback);


Ditto for new line rules. Could you make a pass over all the newlines and see if they can be improved?

guozhangwang · 2017-07-27T21:39:04Z

streams/src/test/java/org/apache/kafka/test/MockProcessorContext.java

        final StateRestoreCallback restoreCallback = restoreFuncs.get(storeName);
-        for (final KeyValue<byte[], byte[]> entry : changeLog) {
-            restoreCallback.restore(entry.key, entry.value);
+        final StateRestoreListener restoreListener = (restoreCallback instanceof StateRestoreListener) ?


We can use the WrappedBatchingStateRestoreCallback here?

guozhangwang · 2017-07-27T21:40:35Z

streams/src/test/java/org/apache/kafka/test/MockProcessorContext.java

@@ -305,4 +328,12 @@ public ProcessorNode currentNode() {
    public void close() {
        metrics.close();
    }
+
+    private static final class NoOpRestoreListener extends AbstractNotifyingBatchingRestoreCallback {


org.apache.kafka.streams.processor.internals.NoOpStateRestoreListener can be used here?

bbejeck · 2017-07-27T23:29:43Z

@guozhangwang @mjsax updates per comments

asfgit · 2017-07-28T00:27:54Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/6402/
Test PASSed (JDK 7 and Scala 2.11).

asfgit · 2017-07-28T00:55:55Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/6387/
Test PASSed (JDK 8 and Scala 2.12).

guozhangwang · 2017-07-28T18:31:26Z

Merged to trunk. THanks @bbejeck !

bbejeck changed the title ~~KAFKA-5363 [WIP] : Initial cut at implementing bulk load for persistent stat…~~ KAFKA-5363 [WIP] : KIP-167 Initial cut at implementing bulk load, restoration event notification Jun 13, 2017

bbejeck force-pushed the KAFKA-5363_add_ability_to_batch_restore branch from 9170d0a to 0f3b947 Compare June 15, 2017 20:42

bbejeck force-pushed the KAFKA-5363_add_ability_to_batch_restore branch from 7bc4575 to 41a21af Compare June 29, 2017 19:43

mjsax reviewed Jul 26, 2017

View reviewed changes

bbejeck added 13 commits July 26, 2017 17:11

KAFKA-5363: Initial cut at adding bulk load for persistent state stor…

5123cf4

…es, adding state restore listener KAFKA-5363: updates per comments on KIP

KAFKA-5363: updates per comments on KIP

04bd74d

KAFKA-5363: Added new interface for notifying state stores of update …

1456759

…start and update end, updated RocksDB to be closed and opened during restores. Added adapter class for abstracting the callback implementations.

KAFKA-5363: Added abstract adapter class for notification of restore …

9334463

…and StateRestoreCallback

KAFKA-5363: Created adapter class to abstract all listener/callback f…

1febe1c

…unctions

KAFKA-5363: updates for KIP changes removed interface, rebased

a3aea90

updates per comments, added javadoc, tests

d2488c0

update from rebase

e0c0c0a

Changes for updating interface to include TopicPartition parameter

e85d858

updates per comments, added test to RocksDB

f3c22df

javadoc updates per comments, refactored test to include smaller tests

2e67246

updates per comments

f0a4bdb

updates for comments.

9b60972

bbejeck force-pushed the KAFKA-5363_add_ability_to_batch_restore branch from 00d98c0 to 9b60972 Compare July 26, 2017 21:22

guozhangwang reviewed Jul 27, 2017

View reviewed changes

bbejeck added 2 commits July 27, 2017 19:27

updates for comments, simplified some code.

c8bc005

updates for comments, simplified some code (all files this time)

ee58a11

asfgit closed this in c50c941 Jul 28, 2017

		}


		private static final class NoOpStateRestoreListener implements StateRestoreListener {


		public class MockStateRestoreListener extends AbstractNotifyingRestoreCallback {

		//Verifies store name called for each state

KAFKA-5363: KIP-167 implementing bulk load, restoration event notification #3325

KAFKA-5363: KIP-167 implementing bulk load, restoration event notification #3325

Conversation

bbejeck commented Jun 13, 2017 • edited

bbejeck commented Jun 13, 2017 • edited

asfbot commented Jun 13, 2017

bbejeck commented Jun 13, 2017

asfbot commented Jun 13, 2017

asfbot commented Jun 13, 2017

bbejeck commented Jun 13, 2017

asfbot commented Jun 13, 2017

asfbot commented Jun 13, 2017

asfbot commented Jun 14, 2017

bbejeck commented Jun 14, 2017 • edited

asfbot commented Jun 14, 2017

bbejeck commented Jun 14, 2017

enothereska commented Jun 14, 2017

asfbot commented Jun 14, 2017

bbejeck commented Jun 14, 2017

asfbot commented Jun 14, 2017

bbejeck commented Jun 14, 2017

asfbot commented Jun 14, 2017

asfbot commented Jun 14, 2017

asfgit commented Jun 15, 2017

asfgit commented Jun 15, 2017

bbejeck commented Jun 20, 2017 • edited

asfgit commented Jun 20, 2017

asfgit commented Jun 20, 2017

asfgit commented Jun 29, 2017

asfgit commented Jun 29, 2017

bbejeck commented Jun 30, 2017

mjsax 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

bbejeck commented Jul 26, 2017

bbejeck commented Jul 26, 2017

asfgit commented Jul 26, 2017

asfgit commented Jul 26, 2017

asfgit commented Jul 26, 2017

asfgit commented Jul 26, 2017

bbejeck commented Jul 27, 2017

guozhangwang 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

bbejeck Jul 27, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bbejeck commented Jul 27, 2017

asfgit commented Jul 28, 2017

asfgit commented Jul 28, 2017

guozhangwang commented Jul 28, 2017

bbejeck commented Jun 13, 2017 •

edited

bbejeck commented Jun 13, 2017 •

edited

bbejeck commented Jun 14, 2017 •

edited

bbejeck commented Jun 20, 2017 •

edited

bbejeck Jul 27, 2017 •

edited