Skip to content

HOTFIX: Fix wrong setting of Serde in MeteredTimestampWindowStore #6808

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
May 28, 2019
Merged

HOTFIX: Fix wrong setting of Serde in MeteredTimestampWindowStore #6808

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
531e984
HOTFIX: Fix wrong setting of Serde in `MeteredTimestampWindowStore`
cadonna May 24, 2019
0501246
Re-throw exception if the `StreamsException` was not caused by `Class…
cadonna May 24, 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
2 changes: 1 addition & 1 deletion ...src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedWindowStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,6 @@ void initStoreSerde(final ProcessorContext context) {
serdes = new StateSerdes<>(
ProcessorStateManager.storeChangelogTopic(context.applicationId(), name()),
keySerde == null ? (Serde<K>) context.keySerde() : keySerde,
valueSerde == null ? new ValueAndTimestampSerde<>((Serde<V>) context.keySerde()) : valueSerde);
valueSerde == null ? new ValueAndTimestampSerde<>((Serde<V>) context.valueSerde()) : valueSerde);
}
}
50 changes: 50 additions & 0 deletions ...c/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampWindowStoreTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,9 @@
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
import org.apache.kafka.streams.state.ValueAndTimestamp;
import org.apache.kafka.streams.state.WindowStore;
import org.apache.kafka.test.InternalMockProcessorContext;
import org.apache.kafka.test.NoOpRecordCollector;
Expand All @@ -35,6 +37,7 @@
import org.junit.Test;

import static org.junit.Assert.assertNull;
import static org.junit.Assert.fail;

public class MeteredTimestampWindowStoreTest {
private InternalMockProcessorContext context;
Expand Down Expand Up @@ -89,4 +92,51 @@ public void shouldNotExceptionIfFetchReturnsNull() {
assertNull(store.fetch("a", 0));
}

@Test
public void shouldNotThrowExceptionIfSerdesCorrectlySetFromProcessorContext() {

@mjsax mjsax May 24, 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.

Can we add those test for MeteredTimestampedKeyValueStoreTest, too. We had the same bug in MeteredTimestampedKeyValueStore and fixed it "on the side" but never added a test...

@cadonna cadonna May 24, 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.

Will open a separate PR for that.

EasyMock.expect(innerStoreMock.name()).andStubReturn("mocked-store");
EasyMock.replay(innerStoreMock);
final MeteredTimestampedWindowStore<String, Long> store = new MeteredTimestampedWindowStore<>(
innerStoreMock,
10L, // any size
"scope",
new MockTime(),
null,
null
);
store.init(context, innerStoreMock);

try {
store.put("key", ValueAndTimestamp.make(42L, 60000));
} catch (final StreamsException exception) {

@mjsax mjsax May 24, 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.

I don't understand this ? Why do we catch StreamsException? Should the test not just fail if an exception is thrown?

@cadonna cadonna May 24, 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.

To distinguish between a test failure and a test that is terminated by an exception. If a StreamsException that caused by a ClassCastException was thrown by store.put(), then the test failed because the behaviour under test was not correct, otherwise it was terminated due to some other issues.

if (exception.getCause() instanceof ClassCastException) {

@bbejeck bbejeck May 24, 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.

nit: IMHO we should just get rid of the try/catch as if the cause is anything other than a ClassCastException this test will still pass.

@cadonna cadonna May 24, 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 will add an else branch and re-throw the exception. That will still distinguish between a failed and a terminated test as explained above.

@mjsax mjsax May 24, 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.

I agree with @bbejeck (compare my other comment about "why we catch StreamsException) -- if any exception is thrown, the test should fail. We don't gain anything IMHO by calling fail vs re-throwing -- it's just confusing the read the test.

@cadonna cadonna May 27, 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.

OK, I see that the tests fail in any case when an exception is thrown. However, I still think that catching StreamException and checking if ClassCastException is its cause is a good idea to document what this test is verifying. Without the try-catch it is just a couple of lines of code without any assert, which may be puzzling when read in a couple of month. Additionally, the distinction between misbehaviour of code under test (i.e. test fails with predefined message in test) and other issue (some exception was thrown) seems still relevant to me for debugging.

If you both still do not agree, I will remove the try-catch clause for the sake of getting this bugfix in.

fail("Serdes are not correctly set from processor context.");
}
throw exception;
}
}

@Test
public void shouldNotThrowExceptionIfSerdesCorrectlySetFromConstructorParameters() {
EasyMock.expect(innerStoreMock.name()).andStubReturn("mocked-store");
EasyMock.replay(innerStoreMock);
final MeteredTimestampedWindowStore<String, Long> store = new MeteredTimestampedWindowStore<>(
innerStoreMock,
10L, // any size
"scope",
new MockTime(),
Serdes.String(),
new ValueAndTimestampSerde<>(Serdes.Long())
);
store.init(context, innerStoreMock);

try {
store.put("key", ValueAndTimestamp.make(42L, 60000));
} catch (final StreamsException exception) {

@mjsax mjsax May 24, 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.

As above

if (exception.getCause() instanceof ClassCastException) {

@bbejeck bbejeck May 24, 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.

same as above

fail("Serdes are not correctly set from constructor parameters.");
}
throw exception;
}
}
}