You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by tzulitai <gi...@git.apache.org> on 2017/12/21 21:51:12 UTC
[GitHub] flink pull request #5200: [FLINK-8306] [kafka] Introduce KafkaOffsetCommitte...
GitHub user tzulitai opened a pull request:
https://github.com/apache/flink/pull/5200
[FLINK-8306] [kafka] Introduce KafkaOffsetCommitter interface
## What is the purpose of the change
This PR is built upon the reworked `FlinkKafkaConsumerBaseTest` in #5188.
Prior to this PR, offset committing was coupled tightly with the `AbstractFetcher`, making unit tests for offset committing behaviours hard to compose concisely. For example, we had tests that required mocking the offset commit methods on `AbstractFetcher`, while ideally, it would be best that those methods are made final (thus, unable to be mocked) to prevent accidental overrides.
This PR decouples offset committing as a separate service behind a new `KafkaOffsetCommitter` interface. For now, the `AbstractFetcher` is reused as an implementation of this service, so that this PR does not introduce any more change other than introducing a new layer of abstraction.
Unit tests that verify offset committing behaviour now provide a dummy verifiable implementation of the `KafkaOffsetCommitter` (instead of using mocks on `AbstractFetcher`) and test against that.
## Brief change log
- Migrate `AbstractFetcher::commitInternalOffsetsToKafka` method to the newly introduced `KafkaOffsetCommitter` interface.
- Let `AbstractFetcher` implement `KafkaOffsetCommitter`
- In the `FlinkKafkaConsumerBase`, let "offset committing" and "record fetching" be logically separated to be handled by two services, i.e. namely a `KafkaOffsetCommitter` and a `AbstractFetcher`. Physically, the fetcher instance sits behind both service abstractions.
- In `FlinkKafkaConsumerBaseTest`, remove all mocks on `AbstractFetcher::commitInternalOffsetsToKafka`, and test against a `KafkaOffsetCommitter` instead.
- Additional hotfix 2906968 that fixes a stale comment referring to an old `AbstractFetcher` behaviour to avoid confusion.
## Verifying this change
This PR does not add any new functionality. Reworked test also do not affect test coverage.
`FlinkKafkaConsumerBaseTest` verifies all changes.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? n/a
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/tzulitai/flink FLINK-8306
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/5200.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #5200
----
commit cb3f488877b7fb2ab0dfcc5c24fe53035bc765e7
Author: Tzu-Li (Gordon) Tai <tz...@...>
Date: 2017-12-20T00:10:44Z
[FLINK-8296] [kafka] Rework FlinkKafkaConsumerBaseTest to not rely on Java reflection
Reflection was mainly used to inject mocks into private fields of the
FlinkKafkaConsumerBase, without the need to fully execute all operator
life cycle methods. This, however, caused the unit tests to be too
implementation-specific.
This commit reworks the FlinkKafkaConsumerBaseTest to remove test
consumer instantiation methods that rely on reflection for dependency
injection. All tests now instantiate dummy test consumers normally, and
let all tests properly execute all operator life cycle methods
regardless of the tested logic.
commit 21491d567ef5d3b8294deec2890b48900c22dd56
Author: Nico Kruber <ni...@...>
Date: 2017-12-19T17:14:19Z
[FLINK-8295] [cassandra] [build] Properly shade netty for the datastax driver
com.datastax.driver.core.NettyUtil expects netty to be present either at its
original package or relocated to com.datastax.shaded.netty. By relocating it
to this package we make sure the driver follows its designated path.
This closes #5183.
commit ef40aaa9d942cc49c3d51816eacdaa5e7dbe9fa5
Author: Tzu-Li (Gordon) Tai <tz...@...>
Date: 2017-12-19T19:55:16Z
[FLINK-8306] [kafka] Introduce KafkaOffsetCommitter interface
Prior to this commit, offset committing was coupled tightly with the
AbstractFetcher, making unit tests for offset committing behaviours hard
to compose concisely. For example, we had tests that mock the offset
commit methods on AbstractFetcher, while ideally, it would be best that
those methods are made final to prevent accidental overrides.
This commit decouples offset committing as a separate service behind a
new KafkaOffsetCommitter interface. For now, the AbstractFetcher is
reused as an implementation of this service.
Unit tests that verify offset committing behaviour now provide a dummy
verifyable implementation of the KafkaOffsetCommitter, instead of using
mocks on AbstractFetcher.
commit 59917867ed19bf9f1ef9b6fba696983675e2747e
Author: Tzu-Li (Gordon) Tai <tz...@...>
Date: 2017-12-20T19:54:40Z
[hotfix] [kafka] Remove stale comment on publishing procedures of AbstractFetcher
The previous comment mentioned "only now will the fetcher return at
least the restored offsets when calling snapshotCurrentState()". This is
a remnant of the previous fetcher initialization behaviour, where in the
past the fetcher wasn't directly seeded with restored offsets on
instantiation.
Since this is no longer true, this commit fixes the stale comment to
avoid confusion.
----
---
[GitHub] flink pull request #5200: [FLINK-8306] [kafka] Introduce KafkaOffsetCommitte...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/5200#discussion_r160103116
--- Diff: flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java ---
@@ -887,4 +894,9 @@ OffsetCommitMode getOffsetCommitMode() {
LinkedMap getPendingOffsetsToCommit() {
return pendingOffsetsToCommit;
}
+
+ @VisibleForTesting
+ KafkaOffsetCommitter createOffsetCommitter() {
--- End diff --
rename to `getOffsetCommitter()`?
---
[GitHub] flink pull request #5200: [FLINK-8306] [kafka] Introduce KafkaOffsetCommitte...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/5200#discussion_r160105955
--- Diff: flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcher.java ---
@@ -51,7 +50,7 @@
* the Flink data streams.
* @param <KPH> The type of topic/partition identifier used by Kafka in the specific version.
*/
-public abstract class AbstractFetcher<T, KPH> {
+public abstract class AbstractFetcher<T, KPH> implements KafkaOffsetCommitter {
--- End diff --
Maybe use composition instead of inheritance here? `AbstractFetcher` as "being a" `KafkaOffsetCommitter` do not make sense to me when I say it out laud, while `AbstractFetcher` "having a" `KafkaOffsetCommitter` sounds better.
---
[GitHub] flink pull request #5200: [FLINK-8306] [kafka] Introduce KafkaOffsetCommitte...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/5200#discussion_r160107329
--- Diff: flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java ---
@@ -89,30 +90,44 @@
@SuppressWarnings("unchecked")
public void testEitherWatermarkExtractor() {
try {
- new DummyFlinkKafkaConsumer<String>(mock(AbstractFetcher.class), mock(AbstractPartitionDiscoverer.class), false)
+ new DummyFlinkKafkaConsumer<String>(
+ mock(AbstractFetcher.class),
+ mock(AbstractPartitionDiscoverer.class),
+ mock(KafkaOffsetCommitter.class),
+ false)
.assignTimestampsAndWatermarks((AssignerWithPeriodicWatermarks<String>) null);
fail();
} catch (NullPointerException ignored) {}
try {
- new DummyFlinkKafkaConsumer<String>(mock(AbstractFetcher.class), mock(AbstractPartitionDiscoverer.class), false)
+ new DummyFlinkKafkaConsumer<String>(
+ mock(AbstractFetcher.class),
+ mock(AbstractPartitionDiscoverer.class),
+ mock(KafkaOffsetCommitter.class),
+ false)
.assignTimestampsAndWatermarks((AssignerWithPunctuatedWatermarks<String>) null);
fail();
} catch (NullPointerException ignored) {}
final AssignerWithPeriodicWatermarks<String> periodicAssigner = mock(AssignerWithPeriodicWatermarks.class);
final AssignerWithPunctuatedWatermarks<String> punctuatedAssigner = mock(AssignerWithPunctuatedWatermarks.class);
- DummyFlinkKafkaConsumer<String> c1 =
- new DummyFlinkKafkaConsumer<>(mock(AbstractFetcher.class), mock(AbstractPartitionDiscoverer.class), false);
+ DummyFlinkKafkaConsumer<String> c1 = new DummyFlinkKafkaConsumer<>(
+ mock(AbstractFetcher.class),
+ mock(AbstractPartitionDiscoverer.class),
+ mock(KafkaOffsetCommitter.class),
--- End diff --
Please replace everywhere `mock(KafkaOffsetCommitter.class)` with `new DummyKafkaOffsetCommitter()` or `new NoOpKafkaOffsetCommitter()`.
---
[GitHub] flink issue #5200: [FLINK-8306] [kafka] Introduce KafkaOffsetCommitter inter...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on the issue:
https://github.com/apache/flink/pull/5200
Subsumed be re-opened PR: #5284
---
[GitHub] flink issue #5200: [FLINK-8306] [kafka] Introduce KafkaOffsetCommitter inter...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on the issue:
https://github.com/apache/flink/pull/5200
@pnowojski thanks a lot for your insightful review!
regarding the choice of composition or inheritance: actually, in the end I think we should be leaning towards neither of both, and let offset committing and record fetching live as separate components. I've left more detailed comments inline.
---
[GitHub] flink pull request #5200: [FLINK-8306] [kafka] Introduce KafkaOffsetCommitte...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/5200#discussion_r160599433
--- Diff: flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcher.java ---
@@ -51,7 +50,7 @@
* the Flink data streams.
* @param <KPH> The type of topic/partition identifier used by Kafka in the specific version.
*/
-public abstract class AbstractFetcher<T, KPH> {
+public abstract class AbstractFetcher<T, KPH> implements KafkaOffsetCommitter {
--- End diff --
The result could also very well be that we should use this opportunity to refactor the vague dependencies between fetcher / consumer thread / consumer base, and include in this PR. I would not be against that.
---
[GitHub] flink issue #5200: [FLINK-8306] [kafka] Introduce KafkaOffsetCommitter inter...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on the issue:
https://github.com/apache/flink/pull/5200
Regarding the race condition you mentioned:
hmm, I can't seem to exactly nail down the "and only first `successfulCommits.inc()` can be omitted because of that" case you mentioned, could you elaborate on that a bit more?
But yes, it seems like there certainly is a race condition here, and can even cause an NPE on:
https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/KafkaConsumerThread.java#L496
Seems like the `nextOffsetsToCommit` and its callback instance should be bundled as a single `AtomicReference` here ...
Would you like to open a PR for that? I wanted to ask because you discovered it in the first place; if not I'll open a fix for it.
---
[GitHub] flink pull request #5200: [FLINK-8306] [kafka] Introduce KafkaOffsetCommitte...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/5200#discussion_r160598939
--- Diff: flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBaseTest.java ---
@@ -89,30 +90,44 @@
@SuppressWarnings("unchecked")
public void testEitherWatermarkExtractor() {
try {
- new DummyFlinkKafkaConsumer<String>(mock(AbstractFetcher.class), mock(AbstractPartitionDiscoverer.class), false)
+ new DummyFlinkKafkaConsumer<String>(
+ mock(AbstractFetcher.class),
+ mock(AbstractPartitionDiscoverer.class),
+ mock(KafkaOffsetCommitter.class),
+ false)
.assignTimestampsAndWatermarks((AssignerWithPeriodicWatermarks<String>) null);
fail();
} catch (NullPointerException ignored) {}
try {
- new DummyFlinkKafkaConsumer<String>(mock(AbstractFetcher.class), mock(AbstractPartitionDiscoverer.class), false)
+ new DummyFlinkKafkaConsumer<String>(
+ mock(AbstractFetcher.class),
+ mock(AbstractPartitionDiscoverer.class),
+ mock(KafkaOffsetCommitter.class),
+ false)
.assignTimestampsAndWatermarks((AssignerWithPunctuatedWatermarks<String>) null);
fail();
} catch (NullPointerException ignored) {}
final AssignerWithPeriodicWatermarks<String> periodicAssigner = mock(AssignerWithPeriodicWatermarks.class);
final AssignerWithPunctuatedWatermarks<String> punctuatedAssigner = mock(AssignerWithPunctuatedWatermarks.class);
- DummyFlinkKafkaConsumer<String> c1 =
- new DummyFlinkKafkaConsumer<>(mock(AbstractFetcher.class), mock(AbstractPartitionDiscoverer.class), false);
+ DummyFlinkKafkaConsumer<String> c1 = new DummyFlinkKafkaConsumer<>(
+ mock(AbstractFetcher.class),
+ mock(AbstractPartitionDiscoverer.class),
+ mock(KafkaOffsetCommitter.class),
--- End diff --
Will do.
---
[GitHub] flink pull request #5200: [FLINK-8306] [kafka] Introduce KafkaOffsetCommitte...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/5200#discussion_r160598888
--- Diff: flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java ---
@@ -559,6 +565,7 @@ public void onException(Throwable cause) {
// the fetchers 'snapshotCurrentState()' method return at least
// the restored offsets
this.kafkaFetcher = fetcher;
+ this.kafkaOffsetCommitter = createOffsetCommitter();
--- End diff --
This is a very valid argument. Will address this with a factory perhaps, as you suggested.
---
[GitHub] flink pull request #5200: [FLINK-8306] [kafka] Introduce KafkaOffsetCommitte...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/5200#discussion_r160107827
--- Diff: flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java ---
@@ -770,8 +777,8 @@ public final void notifyCheckpointComplete(long checkpointId) throws Exception {
return;
}
- final AbstractFetcher<?, ?> fetcher = this.kafkaFetcher;
- if (fetcher == null) {
+ final KafkaOffsetCommitter offsetCommitter = this.kafkaOffsetCommitter;
--- End diff --
why do we need this `final` local variable? `kafkaOffsetCommitter` seems to be "write-once" variable.
---
[GitHub] flink pull request #5200: [FLINK-8306] [kafka] Introduce KafkaOffsetCommitte...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/5200#discussion_r160692827
--- Diff: flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcher.java ---
@@ -51,7 +50,7 @@
* the Flink data streams.
* @param <KPH> The type of topic/partition identifier used by Kafka in the specific version.
*/
-public abstract class AbstractFetcher<T, KPH> {
+public abstract class AbstractFetcher<T, KPH> implements KafkaOffsetCommitter {
--- End diff --
Probably you are right, however I haven't thought this through, but if you had and you feel that's a good opportunity to make some larger refactor I'm fine with that :)
Anyway, as we discussed offline, such refactor could be made in separate PR and here you could just implement `AbstractFetcher#doCommitInternalOffsetsToKafka` for the testing purposes.
I don't mind if you choose one over the other approach :)
---
[GitHub] flink issue #5200: [FLINK-8306] [kafka] Introduce KafkaOffsetCommitter inter...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on the issue:
https://github.com/apache/flink/pull/5200
Regarding the race condition. I think we meant the same thing or maybe I misunderstood part of the code, but yes, this NPE seems like a possible outcome of this race condition.
If you have spare cycles to fix this race feel free to do so, I would like to avoid working on it in this and the following week because of some code handing over before some lengthy vacation in my team.
---
[GitHub] flink pull request #5200: [FLINK-8306] [kafka] Introduce KafkaOffsetCommitte...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai closed the pull request at:
https://github.com/apache/flink/pull/5200
---
[GitHub] flink pull request #5200: [FLINK-8306] [kafka] Introduce KafkaOffsetCommitte...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/5200#discussion_r160598772
--- Diff: flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcher.java ---
@@ -51,7 +50,7 @@
* the Flink data streams.
* @param <KPH> The type of topic/partition identifier used by Kafka in the specific version.
*/
-public abstract class AbstractFetcher<T, KPH> {
+public abstract class AbstractFetcher<T, KPH> implements KafkaOffsetCommitter {
--- End diff --
I agree that composition suits better here, or maybe even neither of both. However, the reality is that currently the offset committing logic is implemented tightly as part of the `AbstractFetcher`, sharing the same Kafka client for both fetching records and committing offsets. Decoupling that would require further refactoring, which I think is a bit out of scope for the current issue at hand.
I have been thinking that we should simply have two separate service implementations for offset committing and record fetching. If that happens, then neither composition or inheritance is required; offset committing and record fetching simply lives as two separate services.
What do you think?
---
[GitHub] flink issue #5200: [FLINK-8306] [kafka] Introduce KafkaOffsetCommitter inter...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on the issue:
https://github.com/apache/flink/pull/5200
As discussed offline:
Will re-open a PR for this that avoids introducing a new interface. I agree that it would make more sense to introduce new interface when we actually decide to go for a major refactor ;)
---
[GitHub] flink issue #5200: [FLINK-8306] [kafka] Introduce KafkaOffsetCommitter inter...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on the issue:
https://github.com/apache/flink/pull/5200
Btw, doesn't the `org.apache.flink.streaming.connectors.kafka.internal.KafkaConsumerThread#setOffsetsToCommit` have a race condition on those lines:
```
// record the work to be committed by the main consumer thread and make sure the consumer notices that
if (nextOffsetsToCommit.getAndSet(offsetsToCommit) != null) {
log.warn("Committing offsets to Kafka takes longer than the checkpoint interval. " +
"Skipping commit of previous offsets because newer complete checkpoint offsets are available. " +
"This does not compromise Flink's checkpoint integrity.");
}
this.offsetCommitCallback = commitCallback;
```
(`getAndSet` followed by `volatile`) with `org.apache.flink.streaming.connectors.kafka.internal.KafkaConsumerThread#run`? However currently it doesn't seem to be so important, since:
1. it's used only for metrics
2. and only first `successfulCommits.inc()` can be omitted because of that., because (at least now) `offsetCommitCallback` is a `write-once` variable, and any subsequent overwrites, are overwriting it to the same value.
---
[GitHub] flink pull request #5200: [FLINK-8306] [kafka] Introduce KafkaOffsetCommitte...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/5200#discussion_r160112834
--- Diff: flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java ---
@@ -559,6 +565,7 @@ public void onException(Throwable cause) {
// the fetchers 'snapshotCurrentState()' method return at least
// the restored offsets
this.kafkaFetcher = fetcher;
+ this.kafkaOffsetCommitter = createOffsetCommitter();
--- End diff --
Changing order of those lines:
```
this.kafkaFetcher = fetcher;
this.kafkaOffsetCommitter = createOffsetCommitter();
```
will brake the code. This is super fragile and is kind of hard to follow.
---