KAFKA-6538: Changes to enhance ByteStore exceptions thrown from RocksDBStore with more human readable info #5103

jadireddi · 2018-05-31T03:33:35Z

https://issues.apache.org/jira/browse/KAFKA-6538

Enhanced exceptions thrown from RocksDBStore with corresponding information for which key/value the operation failed in the wrapping stores (KeyValueStore, WindowedStored, and SessionStore).

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

jadireddi · 2018-05-31T06:49:21Z

@mjsax, Could you please review code changes and let me know if any changes needed.

mjsax

Thanks for the PR! Couple of comment.

mjsax · 2018-06-01T05:33:49Z

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

@@ -223,7 +223,7 @@ private void validateStoreOpen() {
        try {
            return this.db.get(rawKey);
        } catch (final RocksDBException e) {
-            throw new ProcessorStateException("Error while getting value for key from store " + this.name, e);
+            throw new ProcessorStateException("Error while getting value for key %s from store " + this.name, e);


Need to parse this twice before I understood this. I think, we need to add a comment here -- otherwise, we might remove %s later as no formatter is used here and thus it's unclear why the markup is there.

mjsax · 2018-06-01T05:34:37Z

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

@@ -299,13 +299,13 @@ private void putInternal(final byte[] rawKey,
            try {
                db.delete(wOptions, rawKey);
            } catch (final RocksDBException e) {
-                throw new ProcessorStateException("Error while removing key from store " + this.name, e);
+                throw new ProcessorStateException("Error while removing key %s from store " + this.name, e);


mjsax · 2018-06-01T05:34:43Z

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

            }
        } else {
            try {
                db.put(wOptions, rawKey, rawValue);
            } catch (final RocksDBException e) {
-                throw new ProcessorStateException("Error while executing putting key/value into store " + this.name, e);
+                throw new ProcessorStateException("Error while putting key %s value %s into store " + this.name, e);


mjsax · 2018-06-01T05:36:24Z

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

@@ -104,6 +105,10 @@ private void putAndMaybeForward(final ThreadCache.DirtyEntry entry, final Intern
            } else {
                underlying.put(entry.key(), entry.newValue());
            }
+        } catch (final ProcessorStateException e) {


Thinking about this, I am wondering if caching stores are the best place to fix this. Caching might be disabled and thus, the fix would not cover this case?

@mjsax , Just trying to understand. On the point Caching might be disabled, do you mean kafka wouldn't be supporting this it in future or User might have config option to disable Caching?.

@mjsax And does it applies to other Caching files too CachingSessionStore.java and CachingWindowStore.java?

I refer to use configs. Users can disable caching. And yes it, also applies to the other caching stores. I think it would be better to use MeteredXXX stores -- those are always used as wrappers and thus fix the issues.

mjsax · 2018-06-01T05:38:17Z

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

+            bytesStore.put(Bytes.wrap(SessionKeySchema.toBinary(sessionKey, serdes.keySerializer(), topic)), serdes.rawValue(aggregate));
+        } catch (final ProcessorStateException e) {
+            final String message = String.format(e.getMessage(), sessionKey,
+                    aggregate);


nit: move to line above.

mjsax · 2018-06-01T20:37:48Z

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

@@ -223,7 +223,7 @@ private void validateStoreOpen() {
        try {
            return this.db.get(rawKey);
        } catch (final RocksDBException e) {
-            throw new ProcessorStateException("Error while getting value for key from store " + this.name, e);
+            throw new ProcessorStateException("Error while getting value for key %s from store " + this.name, e); // String format is happening in wrapping stores. So formatted message is thrown from wrapping stores.


nit: put the comment in the line above to avoid too long lines :) (same below)

jadireddi · 2018-06-02T09:01:10Z

retest this please

mjsax

Thanks. One follow up question (with cc @guozhangwang )

I am wondering, if we should add some tests for this?

mjsax · 2018-06-05T18:30:37Z

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

-        bytesStore.remove(Bytes.wrap(SessionKeySchema.toBinary(key, serdes.keySerializer(), topic)));
+        try {
+            bytesStore.remove(Bytes.wrap(SessionKeySchema.toBinary(key, serdes.keySerializer(), topic)));
+        }  catch (final ProcessorStateException e) {


Do we need this change here? RocksDBSessionStore should be wrapped with MeteredSessionStore. As an alternative, we can remove the code from MeteredSessionStore thought. Not sure, which class is better for the fix.

\cc @guozhangwang WDYT?

Since all stores will be at least wrapped with MeteredXXStore we can add the error handling only at the metered store layer to be consistent. Besides, the inner RocksDBSessionStore would actually be <Bytes, byte[]> typed always, so this would not help.

Same for the RocksDBWindowStorebelow.

mjsax · 2018-06-05T18:31:47Z

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

-        bytesStore.put(WindowKeySchema.toStoreKeyBinary(key, timestamp, seqnum, serdes), serdes.rawValue(value));
+        try {
+            bytesStore.put(WindowKeySchema.toStoreKeyBinary(key, timestamp, seqnum, serdes), serdes.rawValue(value));
+        } catch (final ProcessorStateException e) {


As above. Not sure if we need this, as the store should be wrapped with MeteredWindowStore.

guozhangwang

The change LGTM.

Just note that we cannot yet close KAFKA-6538 since we have not yet fixed the RecordCollector case where the sent bytes have failed. Maybe worth targeting in another PR though as we are approaching the release deadline. cc @mjsax

guozhangwang · 2018-06-05T18:58:44Z

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

-        bytesStore.remove(Bytes.wrap(SessionKeySchema.toBinary(key, serdes.keySerializer(), topic)));
+        try {
+            bytesStore.remove(Bytes.wrap(SessionKeySchema.toBinary(key, serdes.keySerializer(), topic)));
+        }  catch (final ProcessorStateException e) {


Since all stores will be at least wrapped with MeteredXXStore we can add the error handling only at the metered store layer to be consistent. Besides, the inner RocksDBSessionStore would actually be <Bytes, byte[]> typed always, so this would not help.

Same for the RocksDBWindowStorebelow.

mjsax · 2018-06-05T20:23:46Z

Thanks @guozhangwang! Doing a follow up PR seems reasonable to me. We should also check, if we can remove the generics from RocksDBSessionStore if it's typed <Byte,byte[]> only anyway.

guozhangwang · 2018-06-05T21:19:55Z

Ack. Will leave to @mjsax to merge.

…DBStore with more human readable info.

This reverts commit b07b059580fa1946ffd90e242239eaf9e2468d35.

…XX store classes.

mjsax · 2018-06-06T20:00:30Z

@jadireddi Why did you not add it to the MeteredXXX stores as suggested by Guozhang?

jadireddi · 2018-06-07T01:19:37Z

@mjsax , Added error handling in MeteredSessionStore , MeteredWindowsStore .
As MeteredKeyValueStore and MeteredKeyValueBytesStore using InnerMeteredKeyValueStore to put/get/delete operations. So made changes in InnerMeteredKeyValueStore. Can you please let me know if i am missing something.

mjsax · 2018-06-07T03:41:12Z

@jadireddi Thanks for the explanation. You are right!

…DBStore with more human readable info (#5103) Reviewers: Matthias J. Sax <[email protected]>, Guozhang Wang <[email protected]>

mjsax · 2018-06-07T03:44:42Z

Merged to trunk and cherry-picked to 2.0.

Thanks for the patch @jadireddi!

mjsax · 2018-06-07T03:53:51Z

I closed KAFKA-6538 and created https://issues.apache.org/jira/browse/KAFKA-7015 to track the RecordCollectorImpl issue -- this makes it easier to assign fixes to different versions.

…DBStore with more human readable info (apache#5103) Reviewers: Matthias J. Sax <[email protected]>, Guozhang Wang <[email protected]>

jadireddi mentioned this pull request May 31, 2018

KAFKA-6538: Fixed RocksDBStore to enhance processor state exceptions with more context #4846

Closed

3 tasks

mjsax reviewed Jun 1, 2018

View reviewed changes

mjsax added the streams label Jun 5, 2018

mjsax reviewed Jun 5, 2018

View reviewed changes

guozhangwang approved these changes Jun 5, 2018

View reviewed changes

jadireddi added 9 commits June 6, 2018 17:41

KAFKA-6538: Changes to enhance ByteStore exceptions thrown from Rocks…

f947d24

…DBStore with more human readable info.

KAFKA-6539: Fixed check style issue

5d6583e

Revert "KAFKA-6539: Fixed check style issue"

dfd03f4

This reverts commit b07b059580fa1946ffd90e242239eaf9e2468d35.

KAFKA-6538: Fixed check style issue.

60476d9

KAFKA-6538: Fixed review comments.

c42bc1b

KAFKA-6538: Fixed review comments by doing message format in MeteredX…

cc06f0e

…XX store classes.

KAFKA-6538: Removed unused import.

6b76abd

KAFKA-6538: Code refactored.

854a4df

KAFKA-6538: To revert code changes.

8331c26

jadireddi force-pushed the KAFKA-6538;-To-enhance-RocksDBStore-state-exceptions branch from 1cbaf2b to 8331c26 Compare June 6, 2018 12:12

mjsax approved these changes Jun 7, 2018

View reviewed changes

mjsax merged commit 1509679 into apache:trunk Jun 7, 2018

mjsax pushed a commit that referenced this pull request Jun 7, 2018

KAFKA-6538: Changes to enhance ByteStore exceptions thrown from Rocks…

79c57f2

…DBStore with more human readable info (#5103) Reviewers: Matthias J. Sax <[email protected]>, Guozhang Wang <[email protected]>

jadireddi deleted the KAFKA-6538;-To-enhance-RocksDBStore-state-exceptions branch June 7, 2018 05:01

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-6538: Changes to enhance ByteStore exceptions thrown from RocksDBStore with more human readable info #5103

KAFKA-6538: Changes to enhance ByteStore exceptions thrown from RocksDBStore with more human readable info #5103

jadireddi commented May 31, 2018

jadireddi commented May 31, 2018

mjsax left a comment

mjsax Jun 1, 2018

jadireddi Jun 1, 2018

mjsax Jun 1, 2018

jadireddi Jun 1, 2018

mjsax Jun 1, 2018

jadireddi Jun 1, 2018

mjsax Jun 1, 2018

jadireddi Jun 1, 2018 •

edited

Loading

jadireddi Jun 1, 2018

mjsax Jun 1, 2018

jadireddi Jun 2, 2018

mjsax Jun 1, 2018

jadireddi Jun 1, 2018

mjsax Jun 1, 2018

jadireddi Jun 2, 2018

jadireddi commented Jun 2, 2018 •

edited

Loading

mjsax left a comment

mjsax Jun 5, 2018

guozhangwang Jun 5, 2018

jadireddi Jun 6, 2018 •

edited

Loading

mjsax Jun 5, 2018

jadireddi Jun 6, 2018 •

edited

Loading

guozhangwang left a comment

guozhangwang Jun 5, 2018

mjsax commented Jun 5, 2018

guozhangwang commented Jun 5, 2018

mjsax commented Jun 6, 2018

jadireddi commented Jun 7, 2018 •

edited

Loading

mjsax commented Jun 7, 2018

mjsax commented Jun 7, 2018

mjsax commented Jun 7, 2018

KAFKA-6538: Changes to enhance ByteStore exceptions thrown from RocksDBStore with more human readable info #5103

KAFKA-6538: Changes to enhance ByteStore exceptions thrown from RocksDBStore with more human readable info #5103

Conversation

jadireddi commented May 31, 2018

Committer Checklist (excluded from commit message)

jadireddi commented May 31, 2018

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

jadireddi Jun 1, 2018 • edited Loading

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

jadireddi commented Jun 2, 2018 • edited Loading

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

jadireddi Jun 6, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jadireddi Jun 6, 2018 • edited Loading

Choose a reason for hiding this comment

guozhangwang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mjsax commented Jun 5, 2018

guozhangwang commented Jun 5, 2018

mjsax commented Jun 6, 2018

jadireddi commented Jun 7, 2018 • edited Loading

mjsax commented Jun 7, 2018

mjsax commented Jun 7, 2018

mjsax commented Jun 7, 2018

jadireddi Jun 1, 2018 •

edited

Loading

jadireddi commented Jun 2, 2018 •

edited

Loading

jadireddi Jun 6, 2018 •

edited

Loading

jadireddi Jun 6, 2018 •

edited

Loading

jadireddi commented Jun 7, 2018 •

edited

Loading