Skip to content

KAFKA-8731: InMemorySessionStore throws NullPointerException on startup #7132

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
bbejeck merged 2 commits into from
Jul 31, 2019
Merged

KAFKA-8731: InMemorySessionStore throws NullPointerException on startup #7132

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b2bb961
fix bug
ableegoldman Jul 29, 2019
23303c2
use remove in test
ableegoldman Jul 30, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -120,7 +120,15 @@ public void put(final Windowed<Bytes> sessionKey, final byte[] aggregate) {
@Override
public void remove(final Windowed<Bytes> sessionKey) {
final ConcurrentNavigableMap<Bytes, ConcurrentNavigableMap<Long, byte[]>> keyMap = endTimeMap.get(sessionKey.window().end());
if (keyMap == null) {

@mjsax mjsax Jul 29, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How could keyMap be null, ie, why would we call remove()for a non-existing session (at least in the DSL)?

Not saying that the fix does not make sense, I am just wondering if we need an additional fix for the DSL?

@mjsax mjsax Jul 29, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Well. Thinking about it twice, during restore maybe? Tombstone are preserved longer in the changelog, hence, for a deleted session we only have the tombstone. On restore, we would never see and "insert" but only the delete.

Maybe worth double checking in the DSL code anyway.

@ableegoldman ableegoldman Jul 29, 2019

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I assume a tombstone was being restored, where the corresponding put was already cleaned up from the changelog?

@ableegoldman ableegoldman Jul 29, 2019

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Answered yourself quicker than me. But ok, I'll look around regardless

@ableegoldman ableegoldman Jul 30, 2019

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm fairly confident the only affected case is on the restore path when the changelog contains a tombstone that outlived its corresponding "put" -- during normal processing, if the delete arrives after the key has already been cleaned up it will be dropped as a late record anyways. Assuming we aren't accidentally sending double tombstones anywhere.

return;
}

final ConcurrentNavigableMap<Long, byte[]> startTimeMap = keyMap.get(sessionKey.key());
if (startTimeMap == null) {
return;
}

startTimeMap.remove(sessionKey.window().start());

if (startTimeMap.isEmpty()) {
Expand Down
5 changes: 5 additions & 0 deletions streams/src/test/java/org/apache/kafka/streams/state/internals/SessionBytesStoreTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -472,6 +472,11 @@ public void shouldLogAndMeasureExpiredRecords() {
assertThat(messages, hasItem("Skipping record for expired segment."));
}

@Test
public void shouldNotThrowExceptionRemovingNonexistentKey() {
sessionStore.remove(new Windowed<>("a", new SessionWindow(0, 1)));
}

@Test(expected = NullPointerException.class)
public void shouldThrowNullPointerExceptionOnFindSessionsNullKey() {
sessionStore.findSessions(null, 1L, 2L);
Expand Down