KAFKA-7245 (Deprecate WindowStore#put(key, value)) :- The method in t… by omanges · Pull Request #7105 · apache/kafka

omanges · 2019-07-22T17:44:53Z

WindowStore#put(key, value) has no timestamp as a parameter, so it causes inconsistency to identify to which window does the key belong.

The stream module of Kafka has a window state store that stores the aggregated values for a key in a given time frame. The window store is implemented as an interface, this interface has a strange method named put(key, value), this method has does not have a timestamp as a parameter which is important to determine that to which window frame does the key belongs. In this method, the current record timestamp is used for determining the window frame(as specified in the description of the method), this constraint makes WindowStore error-prone. It is also specified in the method description that method with a timestamp parameter should be used which already present in the interface which expects key, value, and start timestamp as well of the window to which the key belongs. Therefore by deprecating (and finally removing) the method put(key, value), we can prevent inconsistency.

As the method is deprecated, the test classes which used them are needed to be changed when later the method is removed. Following is the list of the classes:

WindowStoreFacadeTest.java
WindowBytesStoreTest.java
SimpleBenchmark.java (benchmarking)
RocksDBWindowStoreTest.java
ProcessorContextImplTest.java
MeteredWindowStoreTest.java
InMemoryWindowStoreTest.java
ChangeLoggingWindowBytesStoreTest.java
CachingWindowStoreTest.java
ChangeLoggingTimestampedWindowBytesStoreTest.java

Committer Checklist (excluded from commit message)

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

…he WindowStore#put(key, value) has no timestamp as parameter, so it causes inconsistency to identify to which window does the key belong.

mjsax · 2019-07-23T00:37:41Z

Thanks for the PR @omanges -- seems the code does not compile... Please update your PR to make it compile so we can review it.

mjsax · 2019-07-23T16:45:21Z

Copied from the Jira ticket:

Hi Matthias J. Sax, as the method in the WindowStore interface and other interfaces and classes which implement WindowStore has been annotated with @deprecated, the test cases which use this method are failing in the build. So do the test cases are also needed to be updated?

The build does not run any tests but fails because it does not compile (note, that compiler warning will fail the build):

Task :streams:compileJava
11:33:40 /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.12/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java:332: warning: [dep-ann] deprecated item is not annotated with @deprecated
11:33:40 public void put(final K key,

...

Do you refer to a local run of the tests? If yes, the question is why exactly did the tests fail?

omanges · 2019-07-23T17:09:58Z

warning: [deprecation] put(K,V) in WindowStore has been deprecated

Similar types of warnings are generated. Due to this, the local build is also not able to compile. So does this means we also need to change test cases as well which are using the method which we have tagged as deprecated?

mjsax · 2019-07-23T20:29:07Z

Yes, you need to add @Deprecated (for case if a test implements the interface) or @SuppressWarning("deprecation") (for case the old deprecated API is used).

…he WindowStore#put(key, value) has no timestamp as parameter, so it causes inconsistency to identify to which window does the key belong.

omanges · 2019-07-24T08:39:17Z

I am able to build successfully on local.

mjsax · 2019-07-24T22:12:49Z

Checkstyle failed. Try to execute ./gradlew streams:clean streams:checkstyleMain streams:checkstyleTest -- if you run streams:test checkstyle goals are included automatically.

…he WindowStore#put(key, value) has no timestamp as parameter, so it causes inconsistency to identify to which window does the key belong.

mjsax

Thanks for the update @omanges! Some comments.

mjsax · 2019-07-25T18:20:07Z

+         * to identify the window frame to which the key belongs.
+         * Use WindowStore#put(key, value, timestamp) instead.
+         *
+         */


We don't need JavaDocs here as we inherit JavaDocs by default automatically. (Similar for all other classes that extend the top level interface -- it's sufficient to have JavaDocs at the top level interface.)

mjsax · 2019-07-25T18:20:35Z

     *              if the serialized bytes are also null it is interpreted as deletes
     * @throws NullPointerException if the given key is {@code null}
+     *
+     * @deprecated as timestamp is not provided for the (key, value) pair, this causes inconsistency


nit: for the key-value pair

mjsax · 2019-07-25T18:20:56Z

+     *
+     * @deprecated as timestamp is not provided for the (key, value) pair, this causes inconsistency
+     * to identify the window frame to which the key belongs.
+     * Use WindowStore#put(key, value, timestamp) instead.


Nit: Use {@link #put(Object, Object, long)} instead.

mjsax · 2019-07-25T18:21:56Z

        runGenericBenchmark(streams, "Streams Stateful Performance [records/latency/rec-sec/MB-sec joined]: ", latch);
    }

+    @SuppressWarnings("deprecation")


I think for SimpleBenchmark we should update the code instead of suppressing the warning.

So the method call store.put(key, value), can be replaced by store.put(key, value, timestamp) as we are able to retrieve timestamp from the context. @mjsax what do you think?

I have updated the code in the latest commit please have a look.

mjsax · 2019-07-25T18:26:39Z

    }

    @Test
+    @SuppressWarnings("deprecation")


This this and the other tests in this class: I am not sure what the best approach is, and if we should rather update the code -- or maybe even duplicate the code?

Similar for some other test classes.

\cc @guozhangwang @bbejeck @vvcephei @ableegoldman @cadonna @abbccdda WDYT?

Personally I think we should migrate all these tests to use the non-deprecated put then add a test or two just verifying put(key) and put(key, time) have the same results. But IMO all the tests which aren't explicitly testing the deprecated method should be using the "correct" one

I'd agree with @ableegoldman : for test coverage of functionality, we should touch on non-deprecated code only; and only having a few deprecated test that tests the deprecated methods acts the same behavior as their delegated method.

Thanks! That sounds reasonable to me!

Hi, @mjsax the timestamp is not present in the local context of the test cases. So in order to replace the deprecated method with the new method, we need some timestamp. So can we use the 'DEFAULT_TIMESTAMP' which defined as a global in the class CachingWindowStoreTest.

For shouldNotReturnDuplicatesInRanges it seems best to use processorContext.timestamp() -- processorContext is passed in init() do you just need to add a member variable to the Transformer to store is so you can use it in transform()

Hi, @mjsax I have made changes for the shouldNotReturnDuplicatesInRanges. For the remaining usage i.e for the cachingStore do we need to update to make use of the method with timestamp?

Hi, @mjsax it seems that an exception is thrown if a call is made to access the timestamp in the init.
Exception:-
org.apache.kafka.streams.state.internals.CachingWindowStoreTest > shouldNotReturnDuplicatesInRanges FAILED
org.apache.kafka.streams.errors.StreamsException: failed to initialize processor KSTREAM-TRANSFORM-0000000001
at org.apache.kafka.streams.processor.internals.ProcessorNode.init(ProcessorNode.java:96)
at org.apache.kafka.streams.processor.internals.StreamTask.initTopology(StreamTask.java:526)
at org.apache.kafka.streams.processor.internals.StreamTask.initializeTopology(StreamTask.java:253)
at org.apache.kafka.streams.TopologyTestDriver.(TopologyTestDriver.java:377)
at org.apache.kafka.streams.TopologyTestDriver.(TopologyTestDriver.java:240)
at org.apache.kafka.streams.state.internals.CachingWindowStoreTest.shouldNotReturnDuplicatesInRanges(CachingWindowStoreTest.java:180)

Caused by: java.lang.IllegalStateException: This should not happen as timestamp() should only be called while a record is processed at org.apache.kafka.streams.processor.internals.AbstractProcessorContext.timestamp(AbstractProcessorContext.java:158) at org.apache.kafka.streams.state.internals.CachingWindowStoreTest$1.init(CachingWindowStoreTest.java:134) at org.apache.kafka.streams.kstream.internals.TransformerSupplierAdapter$1.init(TransformerSupplierAdapter.java:42) at org.apache.kafka.streams.kstream.internals.KStreamFlatTransform$KStreamFlatTransformProcessor.init(KStreamFlatTransform.java:51) at org.apache.kafka.streams.processor.internals.ProcessorNode.init(ProcessorNode.java:92)

That is expected. You need to add a member variable ProcessorContext context to the class and assign this.context = context in init().

In transform() you can use context.timestamp() to get the current timestamp.

mjsax · 2019-07-25T18:31:18Z

        });
    }

+


not: avoid unnecessary diffs

…he WindowStore#put(key, value) has no timestamp as parameter, so it causes inconsistency to identify to which window does the key belong.

omanges · 2019-08-16T16:09:17Z

Is Jenkins server down? I am not able to see the cause of the build fail.

ableegoldman · 2019-08-16T16:27:11Z

Build results get removed after a few days. Probably a flaky test, but try running locally just in case.

Retest this, please

omanges · 2019-08-22T12:42:31Z

Can someone help me I am not able to understand the reason for the failure of the test cases?

omanges · 2019-08-22T12:54:57Z

How can I trigger tests again?

ableegoldman · 2019-08-22T19:04:07Z

You can retrigger them by commenting:

retest this, please

omanges · 2019-08-23T05:19:40Z

Hi please retest this.

omanges · 2019-08-23T06:23:24Z

The test is not failing because of the changes I am unable to understand, what’s the problem.

ableegoldman · 2019-08-23T17:09:36Z

Yeah, unfortunately there are a number of flaky tests across kafka projects. They can fail for environmental reasons not related to your PR. If you see a failure you believe is unrelated to your changes, please submit a JIRA for the flaky test (or comment on an existing one to note there was an additional failure). Generally we just comment on the PR saying which builds passed and which test failed on the ones that didn't, then retrigger the tests.
However, I notice one of the failing tests may be due to your changes. Can you look into org.apache.kafka.streams.state.internals.CachingWindowStoreTest.shouldNotReturnDuplicatesInRanges ?

mjsax · 2019-09-04T04:13:18Z

@omanges -- I would recommend to run the test locally first and make sure they pass. If they pass locally and there are test failures on Jenkins, it might be due to test flakiness.

Did you run the tests locally? Also to trigger a Jenkins retest, you need to use the exact phrase retest this please (without comma as Sophie did above; also Hi please retest this. won't work)

…he WindowStore#put(key, value) has no timestamp as parameter, so it causes inconsistency to identify to which window does the key belong.

omanges · 2019-09-13T05:44:08Z

@mjsax I think the first one failed due to some flaky test.

omanges · 2019-09-17T05:45:50Z

So are there any further changes needed for this PR?

ableegoldman

Thanks @omanges! LGTM

Implements KIP-474. Reviewers: A. Sophie Blee-Goldman <sophie@confluent.io>, Matthias J. Sax <matthias@confluent.io>

mjsax · 2019-10-07T21:52:05Z

Merged to trunk and cherry-picked to 2.4 branch.

Thanks for the KIP and PR @omanges!

omanges · 2019-10-22T09:54:12Z

Thanks.

…t-for-generated-requests * apache-github/trunk: KAFKA-8932; Add tag for CreateTopicsResponse.TopicConfigErrorCode (KIP-525) (apache#7464) KAFKA-8944: Fixed KTable compiler warning. (apache#7393) KAFKA-8964: Rename tag client-id for thread-level metrics and below (apache#7429) MINOR: remove unused imports in Streams system tests (apache#7468) KAFKA-7190; Retain producer state until transactionalIdExpiration time passes (apache#7388) KAFKA-8983; AdminClient deleteRecords should not fail all partitions unnecessarily (apache#7449) MINOR: Modified Exception handling for KIP-470 (apache#7461) KAFKA-7245: Deprecate WindowStore#put(key, value) (apache#7105) KAFKA-8179: Part 7, cooperative rebalancing in Streams (apache#7386) KAFKA-8985; Add flexible version support to inter-broker APIs (apache#7453) MINOR: Bump version to 2.5.0-SNAPSHOT (apache#7455)

KAFKA-7245 (Deprecate WindowStore#put(key, value)) :- The method in t…

cc5c8ca

…he WindowStore#put(key, value) has no timestamp as parameter, so it causes inconsistency to identify to which window does the key belong.

omanges added 2 commits July 24, 2019 11:48

Merge branch 'trunk' of https://github.com/apache/kafka into trunk

7a3548a

KAFKA-7245 (Deprecate WindowStore#put(key, value)) :- The method in t…

26674d6

…he WindowStore#put(key, value) has no timestamp as parameter, so it causes inconsistency to identify to which window does the key belong.

omanges added 2 commits July 25, 2019 10:53

KAFKA-7245 (Deprecate WindowStore#put(key, value)) :- The method in t…

af40da2

…he WindowStore#put(key, value) has no timestamp as parameter, so it causes inconsistency to identify to which window does the key belong.

Merge branch 'trunk' of https://github.com/apache/kafka into trunk

e4eebb5

mjsax reviewed Jul 25, 2019

View reviewed changes

omanges added 5 commits July 29, 2019 10:42

KAFKA-7245 (Deprecate WindowStore#put(key, value)) :- The method in t…

8d9404d

…he WindowStore#put(key, value) has no timestamp as parameter, so it causes inconsistency to identify to which window does the key belong.

Merge branch 'trunk' of https://github.com/apache/kafka into trunk

477846e

Merge branch 'trunk' of https://github.com/apache/kafka into trunk

f11aee7

KAFKA-7245 (Deprecate WindowStore#put(key, value)) :- The method in t…

0ddd744

…he WindowStore#put(key, value) has no timestamp as parameter, so it causes inconsistency to identify to which window does the key belong.

Merge branch 'trunk' of https://github.com/apache/kafka into trunk

44335ac

Merge branch 'trunk' of https://github.com/apache/kafka into trunk

8798481

Merge branch 'trunk' of https://github.com/apache/kafka into trunk

211edb6

omanges added 2 commits September 9, 2019 14:04

Merge branch 'trunk' of https://github.com/apache/kafka into trunk

88bceda

Merge branch 'trunk' of https://github.com/apache/kafka into trunk

96d6a7d

KAFKA-7245 (Deprecate WindowStore#put(key, value)) :- The method in t…

7a13ea3

…he WindowStore#put(key, value) has no timestamp as parameter, so it causes inconsistency to identify to which window does the key belong.

omanges requested a review from mjsax September 18, 2019 09:50

omanges requested a review from guozhangwang October 5, 2019 09:41

mjsax added the streams label Oct 7, 2019

ableegoldman approved these changes Oct 7, 2019

View reviewed changes

mjsax merged commit cfa1067 into apache:trunk Oct 7, 2019

mjsax pushed a commit that referenced this pull request Oct 7, 2019

KAFKA-7245: Deprecate WindowStore#put(key, value) (#7105)

bd9a177

Implements KIP-474. Reviewers: A. Sophie Blee-Goldman <sophie@confluent.io>, Matthias J. Sax <matthias@confluent.io>

mjsax added the kip Requires or implements a KIP label Jun 12, 2020

Conversation

omanges commented Jul 22, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

mjsax commented Jul 23, 2019

Uh oh!

mjsax commented Jul 23, 2019

Uh oh!

omanges commented Jul 23, 2019

Uh oh!

mjsax commented Jul 23, 2019

Uh oh!

omanges commented Jul 24, 2019

Uh oh!

mjsax commented Jul 24, 2019

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

omanges commented Aug 16, 2019

Uh oh!

ableegoldman commented Aug 16, 2019

Uh oh!

omanges commented Aug 22, 2019

Uh oh!

omanges commented Aug 22, 2019

Uh oh!

ableegoldman commented Aug 22, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

omanges commented Aug 23, 2019

Uh oh!

omanges commented Aug 23, 2019

Uh oh!

ableegoldman commented Aug 23, 2019

Uh oh!

mjsax commented Sep 4, 2019

Uh oh!

omanges commented Sep 13, 2019

Uh oh!

omanges commented Sep 17, 2019

Uh oh!

ableegoldman left a comment

Choose a reason for hiding this comment

Uh oh!

omanges commented Jul 22, 2019 •

edited

Loading

ableegoldman commented Aug 22, 2019 •

edited

Loading