Skip to content
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.

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

KAFKA-8615: Change to track partition time breaks TimestampExtractor #7054

Merged
merged 13 commits into from Jul 18, 2019

Conversation

@ableegoldman
Copy link
Contributor

commented Jul 9, 2019

The timestamp extractor takes a previousTimestamp parameter which should be the partition time. This PR adds back in partition time tracking for the extractor, and renames previousTimestamp --> partitionTime

Should be cherry-picked back to 2.1

@ableegoldman

This comment has been minimized.

Copy link
Contributor Author

commented Jul 9, 2019

@mjsax mjsax added the streams label Jul 9, 2019

@mjsax
Copy link
Member

left a comment

Can we add a test?

@mjsax

This comment has been minimized.

Copy link
Member

commented Jul 9, 2019

Also, I think we need to add back the notion of partition time. Currently, we pass in next() timestamp that is not the previous timestamp (ie, look into the "future" but we should look into the "past")

ableegoldman added 5 commits Jul 9, 2019
@bbejeck
Copy link
Contributor

left a comment

Thanks for the patch @ableegoldman, just one minor comment otherwise lgtm modulo comments from @mjsax.

@bbejeck

This comment has been minimized.

Copy link
Contributor

commented Jul 9, 2019

retest this please

@mjsax
Copy link
Member

left a comment

Some follow up comments, based on our in-person discussion today.

@@ -167,7 +169,7 @@ private void updateHead() {

final long timestamp;
try {
timestamp = timestampExtractor.extract(deserialized, timestamp());
timestamp = timestampExtractor.extract(deserialized, partitionTime);

This comment has been minimized.

Copy link
@mjsax

mjsax Jul 11, 2019

Member

Can we also update method timestamp() to headRecordTimestamp() to be more explicit what it returns? It's orthogonal to the actual fix, but might be a good improvement.

This comment has been minimized.

Copy link
@mjsax

mjsax Jul 11, 2019

Member

Similar, can we piggy-back some cleanup to PartitionGroup ?

- rename PartitionGroup#timestamp() to PartitionGroup#streamTime()
(also update the JavaDocs, that seems to be wrong)

- in `clear()` reset `streamTime` to UNKNOWN ?

- in `nextRecord()`: do we need to check if `queue != null` and do we need to check if `record != null` (seem it's ensure that both can never be `null` ?)

This comment has been minimized.

Copy link
@ableegoldman

ableegoldman Jul 11, 2019

Author Contributor

Ack to all...except the last point. We do check both for null..?

This comment has been minimized.

Copy link
@mjsax

mjsax Jul 11, 2019

Member

Seems, we check both for null atm:

        final RecordQueue queue = nonEmptyQueuesByTime.poll();
        info.queue = queue;

        if (queue != null) {
            // get the first record from this queue.
            record = queue.poll();

            if (record != null) {
                --totalBuffered;

But I think they cannot be null, could they?

This comment has been minimized.

Copy link
@ableegoldman

ableegoldman Jul 11, 2019

Author Contributor

Sorry, misunderstood your question. Yes, either one could potentially be null if we don't yet have new records to process?

This comment has been minimized.

Copy link
@mjsax

mjsax Jul 16, 2019

Member

Good point. final RecordQueue queue = nonEmptyQueuesByTime.poll(); could return null. However, I am wondering if record = queue.poll(); could return null, because it's called nonEmptyQueuesByTime -- hence, queue should never be empty?

This comment has been minimized.

Copy link
@guozhangwang

guozhangwang Jul 18, 2019

Contributor

I think I agree the second null check should never happen.

@@ -41,8 +41,8 @@
*
* PartitionGroup also maintains a stream-time for the group as a whole.

This comment has been minimized.

Copy link
@mjsax

mjsax Jul 11, 2019

Member

Maybe we should change "stream-time" to "task-time" ? @vvcephei suggested it (and I like it) in an in-person discussion? If we agree, also the corresponding variable and method names should be updated. Thoughts?

Should it be "stream-time" or "stream time" ?

This comment has been minimized.

Copy link
@ableegoldman

ableegoldman Jul 11, 2019

Author Contributor

I support "task-time" -- would be good to distinguish from "stream time" should we ever want/need a "global" stream time. That said, we use stream time all over including things like PunctuationType. Will it be confusing to have maybePuncuateStreamTime() punctuate on something called task-time? Though we already refer to it separately as "partition group timestamp", "stream partition time", AND "stream time" in that method..

Regarding "stream-time" vs "stream time" -- we use the hyphen when referring to types of time semantics (eg event-time) so I'd favor "stream time" to maintain a distinction between "semantics types" and "time definitions" -- WDYT?

This comment has been minimized.

Copy link
@ableegoldman

ableegoldman Jul 11, 2019

Author Contributor

On the other hand, maybe phrases like "lowest task time" might be ambiguous (what does "lowest task" mean and why do we care about its time?) so I'm good with task-time.

* The PartitionGroup's stream-time is also the stream-time of its task and is used as the
* stream-time for any computations that require it.
* Note however that any computation that depends on stream time tracks it on a per-operator basis to obtain an
* accurate view of the local stream time as seen by that node.

This comment has been minimized.

Copy link
@mjsax

mjsax Jul 11, 2019

Member

node -> processor? Maybe we could call this "processor time" ?

This comment has been minimized.

Copy link
@ableegoldman

ableegoldman Jul 11, 2019

Author Contributor

How about just local stream time -> local time to avoid introducing too many types of time? Or do you feel "processor time" is clear enough (possibly more so than "local time" -- WDYT?)

This comment has been minimized.

Copy link
@ableegoldman

ableegoldman Jul 11, 2019

Author Contributor

In light of renaming "stream time" -> "task time" I think it does make sense to call this "processor time" and establish a clear naming hierarchy

@@ -167,7 +169,7 @@ private void updateHead() {

final long timestamp;
try {
timestamp = timestampExtractor.extract(deserialized, timestamp());
timestamp = timestampExtractor.extract(deserialized, partitionTime);

This comment has been minimized.

Copy link
@mjsax

mjsax Jul 11, 2019

Member

Seems, we check both for null atm:

        final RecordQueue queue = nonEmptyQueuesByTime.poll();
        info.queue = queue;

        if (queue != null) {
            // get the first record from this queue.
            record = queue.poll();

            if (record != null) {
                --totalBuffered;

But I think they cannot be null, could they?

@bbejeck

This comment has been minimized.

Copy link
Contributor

commented Jul 11, 2019

retest this please

@bbejeck

This comment has been minimized.

Copy link
Contributor

commented Jul 12, 2019

java 11 scala 2.13 timed out, java 8 failure unrelated

retest this please

@bbejeck

This comment has been minimized.

Copy link
Contributor

commented Jul 15, 2019

java 11 scala 2.12 failed, java 8 failed
java 11 scala 2.13 passed All test results cleaned out already

retest this please

@mjsax

This comment has been minimized.

Copy link
Member

commented Jul 16, 2019

Java11/2.12: (https://issues.apache.org/jira/browse/KAFKA-8672)

java.lang.RuntimeException: Could not find enough records. found 33, expected 100
	at org.apache.kafka.connect.util.clusters.EmbeddedKafkaCluster.consume(EmbeddedKafkaCluster.java:306)
	at org.apache.kafka.connect.integration.RebalanceSourceConnectorsIntegrationTest.testReconfigConnector(RebalanceSourceConnectorsIntegrationTest.java:180)

Java11/2.13 (https://issues.apache.org/jira/browse/KAFKA-8555)

org.apache.kafka.connect.errors.DataException: Insufficient records committed by connector simple-conn in 15000 millis. Records expected=2000, actual=1500
	at org.apache.kafka.connect.integration.ConnectorHandle.awaitCommits(ConnectorHandle.java:188)
	at org.apache.kafka.connect.integration.ExampleConnectIntegrationTest.testSourceConnector(ExampleConnectIntegrationTest.java:181)

Java8: env error

stderr: fatal: unable to connect to github.com:
13:08:31 github.com: Temporary failure in name resolution

Retest this please.

@mjsax

This comment has been minimized.

Copy link
Member

commented Jul 16, 2019

Java8 failed again with env error
Java 11/2.13 failed with know ExampleConnectIntegrationTest.testSourceConnector
Java 11/2.12 passed.

Retest this please.

@ableegoldman

This comment has been minimized.

Copy link
Contributor Author

commented Jul 16, 2019

Java8 and 11.12 passed, Java11.13 failed with knownkafka.admin.ResetConsumerGroupOffsetTest.testResetOffsetsExistingTopic and known kafka.server.LogDirFailureTest.testIOExceptionDuringLogRoll

*/
public long timestamp() {
public long streamTime() {

This comment has been minimized.

Copy link
@mjsax

mjsax Jul 16, 2019

Member

Should we rename this this taskTime()? (Just a thought).

This comment has been minimized.

Copy link
@ableegoldman

ableegoldman Jul 17, 2019

Author Contributor

I'm in favor of that in theory -- but, then do we also rename maybePuncuateStreamTime ? Do we also deprecate PunctuationType.STREAM_TIME in favor of PunctuationType.TASK_TIME? Task time does seem more appropriate but I'm hesitant to mix terminology ...

This comment has been minimized.

Copy link
@guozhangwang

guozhangwang Jul 18, 2019

Contributor

Let's just keep it as streamTime for now.

@guozhangwang
Copy link
Contributor

left a comment

LGTM. @ableegoldman I think we can remove the second null check but I'm okay keeping it as well just to be safe.

*/
public long timestamp() {
public long streamTime() {

This comment has been minimized.

Copy link
@guozhangwang

guozhangwang Jul 18, 2019

Contributor

Let's just keep it as streamTime for now.

@@ -167,7 +169,7 @@ private void updateHead() {

final long timestamp;
try {
timestamp = timestampExtractor.extract(deserialized, timestamp());
timestamp = timestampExtractor.extract(deserialized, partitionTime);

This comment has been minimized.

Copy link
@guozhangwang

guozhangwang Jul 18, 2019

Contributor

I think I agree the second null check should never happen.

@guozhangwang guozhangwang merged commit 62fbc92 into apache:trunk Jul 18, 2019

2 of 3 checks passed

JDK 11 and Scala 2.13 FAILURE 11671 tests run, 67 skipped, 2 failed.
Details
JDK 11 and Scala 2.12 SUCCESS 11671 tests run, 67 skipped, 0 failed.
Details
JDK 8 and Scala 2.11 SUCCESS 11671 tests run, 67 skipped, 0 failed.
Details
guozhangwang added a commit that referenced this pull request Jul 18, 2019
KAFKA-8615: Change to track partition time breaks TimestampExtractor (#…
…7054)

The timestamp extractor takes a previousTimestamp parameter which should be the partition time. This PR adds back in partition time tracking for the extractor, and renames previousTimestamp --> partitionTime

Reviewers: Guozhang Wang <wangguoz@gmail.com>, Bill Bejeck <bbejeck@gmail.com>, Matthias J. Sax <mjsax@apache.org>
@guozhangwang

This comment has been minimized.

Copy link
Contributor

commented Jul 18, 2019

Cherry-picked to 2.3 as well.

@ableegoldman

This comment has been minimized.

Copy link
Contributor Author

commented Jul 18, 2019

Thanks @guozhangwang! I think it should actually be cherry-picked all the way back to 2.1

I don't think there should be conflicts but if you need a separate PR for the earlier branches, let me know

guozhangwang added a commit that referenced this pull request Jul 18, 2019
KAFKA-8615: Change to track partition time breaks TimestampExtractor (#…
…7054)

The timestamp extractor takes a previousTimestamp parameter which should be the partition time. This PR adds back in partition time tracking for the extractor, and renames previousTimestamp --> partitionTime

Reviewers: Guozhang Wang <wangguoz@gmail.com>, Bill Bejeck <bbejeck@gmail.com>, Matthias J. Sax <mjsax@apache.org>
guozhangwang added a commit that referenced this pull request Jul 19, 2019
KAFKA-8615: Change to track partition time breaks TimestampExtractor (#…
…7054)

The timestamp extractor takes a previousTimestamp parameter which should be the partition time. This PR adds back in partition time tracking for the extractor, and renames previousTimestamp --> partitionTime

Reviewers: Guozhang Wang <wangguoz@gmail.com>, Bill Bejeck <bbejeck@gmail.com>, Matthias J. Sax <mjsax@apache.org>
@guozhangwang

This comment has been minimized.

Copy link
Contributor

commented Jul 19, 2019

Cherry-picked to 2.2 and 2.1.

ijuma added a commit to confluentinc/kafka that referenced this pull request Jul 20, 2019
Merge remote-tracking branch 'apache-github/2.3' into ccs-2.3
* apache-github/2.3:
  MINOR: Update documentation for enabling optimizations (apache#7099)
  MINOR: Remove stale streams producer retry default docs. (apache#6844)
  KAFKA-8635; Skip client poll in Sender loop when no request is sent (apache#7085)
  KAFKA-8615: Change to track partition time breaks TimestampExtractor (apache#7054)
  KAFKA-8670; Fix exception for kafka-topics.sh --describe without --topic mentioned (apache#7094)
  KAFKA-8602: Separate PR for 2.3 branch (apache#7092)
  KAFKA-8530; Check for topic authorization errors in OffsetFetch response (apache#6928)
  KAFKA-8662; Fix producer metadata error handling and consumer manual assignment (apache#7086)
  KAFKA-8637: WriteBatch objects leak off-heap memory (apache#7050)
  KAFKA-8620: fix NPE due to race condition during shutdown while rebalancing (apache#7021)
  HOT FIX: close RocksDB objects in correct order (apache#7076)
  KAFKA-7157: Fix handling of nulls in TimestampConverter (apache#7070)
  KAFKA-6605: Fix NPE in Flatten when optional Struct is null (apache#5705)
  Fixes #8198 KStreams testing docs use non-existent method pipe (apache#6678)
  KAFKA-5998: fix checkpointableOffsets handling (apache#7030)
  KAFKA-8653; Default rebalance timeout to session timeout for JoinGroup v0 (apache#7072)
  KAFKA-8591; WorkerConfigTransformer NPE on connector configuration reloading (apache#6991)
  MINOR: add upgrade text (apache#7013)
  Bump version to 2.3.1-SNAPSHOT
xiowu0 added a commit to linkedin/kafka that referenced this pull request Aug 22, 2019
[LI-CHERRY-PICK] [3d7b989] KAFKA-8615: Change to track partition time…
… breaks TimestampExtractor (apache#7054)

TICKET = KAFKA-8615
LI_DESCRIPTION =
EXIT_CRITERIA = HASH [3d7b989]
ORIGINAL_DESCRIPTION =

The timestamp extractor takes a previousTimestamp parameter which should be the partition time. This PR adds back in partition time tracking for the extractor, and renames previousTimestamp --> partitionTime

Reviewers: Guozhang Wang <wangguoz@gmail.com>, Bill Bejeck <bbejeck@gmail.com>, Matthias J. Sax <mjsax@apache.org>
(cherry picked from commit 3d7b989)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Projects
None yet
4 participants
You can’t perform that action at this time.