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 2018/01/10 06:23:37 UTC
[GitHub] flink pull request #5269: [FLINK-6004] [kinesis] Allow FlinkKinesisConsumer ...
GitHub user tzulitai opened a pull request:
https://github.com/apache/flink/pull/5269
[FLINK-6004] [kinesis] Allow FlinkKinesisConsumer to skip non-deserializable records
## What is the purpose of the change
This PR is based on #5268, which includes fixes to harden Kinesis unit tests. Only the last commit is relevant.
In the past, we allowed the Flink Kafka Consumer to skip corrupted Kafka records which cannot be deserialized. In reality pipelines, it is entirely normal that this could happen.
This PR adds this functionality to the Flink Kinesis Consumer also.
## Brief change log
- Clarify in Javadoc of `KinesisDeserializationSchema` that `null` can be returned if a message cannot be deserialized.
- If `record` is `null` in `KinesisDataFetcher::emitRecordAndUpdateState(...)`, do not collect any output for the record.
- Add `KinesisDataFetcherTest::testSkipCorruptedRecord()` to verify feature.
## Verifying this change
Additional `KinesisDataFetcherTest::testSkipCorruptedRecord()` test verifies this change.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / **no**)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / **no**)
- The serializers: (yes / **no** / don't know)
- The runtime per-record code paths (performance sensitive): (**yes** / no / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes / **no** / don't know)
- The S3 file system connector: (yes / **no** / don't know)
## Documentation
- Does this pull request introduce a new feature? (**yes** / no)
- If yes, how is the feature documented? (not applicable / docs / **JavaDocs** / not documented)
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/tzulitai/flink FLINK-6004
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/5269.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 #5269
----
commit 94b45919afa5a3ec3ce68c45e57f7989397f9640
Author: Tzu-Li (Gordon) Tai <tz...@...>
Date: 2018-01-10T02:11:31Z
[FLINK-8398] [kinesis, tests] Cleanup confusing implementations in KinesisDataFetcherTest and related classes
The previous implementation of the TestableKinesisDataFetcher was
confusing in various ways, causing it hard to be re-used for other
tests. This commit contains the following various cleaups:
- Remove confusing mocks of source context and checkpoint lock. We now
allow users of the TestableKinesisDataFetcher to provide a source
context, which should provide the checkpoint lock.
- Remove override of emitRecordAndUpdateState(). Strictly speaking, that
method should be final. It was previously overriden to allow
verifying how many records were output by the fetcher. That
verification would be better implemented within a mock source context.
- Properly parameterize the output type for the
TestableKinesisDataFetcher.
- Remove use of PowerMockito in KinesisDataFetcherTest.
- Use CheckedThreads to properly capture any exceptions in fetcher /
consumer threads in unit tests.
- Use assertEquals / assertNull instead of assertTrue where-ever
appropriate.
commit 547d19f9196512231661f427f3792f2e1f831339
Author: Tzu-Li (Gordon) Tai <tz...@...>
Date: 2018-01-10T05:41:49Z
[FLINK-8398] [kinesis, tests] Stabilize flaky KinesisDataFetcherTests
Prior to this commit, several unit tests in KinesisDataFetcherTest
relied on sleeps to wait until a certain operation happens, in order for
the test to pass.
This commit removes those sleeps and replaces the test behaviours with
OneShotLatches.
commit 8d2b086ae7133999ec03620e3434cd659fd8d9d3
Author: Tzu-Li (Gordon) Tai <tz...@...>
Date: 2018-01-10T06:04:10Z
[FLINK-6004] [kinesis] Allow FlinkKinesisConsumer to skip records
This commit acknowledges that null can be returned from the
deserialization schema, if the message cannot be deserialized. If null
is returned for a Kinesis record, no output is produced for that record,
while the sequence number in the shard state is still advanced so that
the record is effectively accounted as processed.
----
---
[GitHub] flink pull request #5269: [FLINK-6004] [kinesis] Allow FlinkKinesisConsumer ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5269#discussion_r165020094
--- Diff: flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/internals/KinesisDataFetcherTest.java ---
@@ -86,6 +88,56 @@ public void testIfNoShardsAreFoundShouldThrowException() throws Exception {
fetcher.runFetcher(); // this should throw RuntimeException
}
+ @Test
+ public void testSkipCorruptedRecord() throws Exception {
+ final String stream = "fakeStream";
+ final int numShards = 3;
+
+ final LinkedList<KinesisStreamShardState> testShardStates = new LinkedList<>();
+ final TestSourceContext<String> sourceContext = new TestSourceContext<>();
+
+ final TestableKinesisDataFetcher<String> fetcher = new TestableKinesisDataFetcher<>(
+ Collections.singletonList(stream),
+ sourceContext,
+ TestUtils.getStandardProperties(),
+ new KinesisDeserializationSchemaWrapper<>(new SimpleStringSchema()),
+ 1,
+ 0,
+ new AtomicReference<>(),
+ testShardStates,
+ new HashMap<>(),
+ FakeKinesisBehavioursFactory.nonReshardedStreamsBehaviour(Collections.singletonMap(stream, numShards)));
+
+ // FlinkKinesisConsumer is responsible for setting up the fetcher before it can be run;
+ // run the consumer until it reaches the point where the fetcher starts to run
+ final DummyFlinkKafkaConsumer<String> consumer = new DummyFlinkKafkaConsumer<>(TestUtils.getStandardProperties(), fetcher, 1, 0);
+
+ CheckedThread consumerThread = new CheckedThread() {
+ @Override
+ public void go() throws Exception {
+ consumer.run(new TestSourceContext<>());
+ }
+ };
+ consumerThread.start();
+
+ fetcher.waitUntilRun();
+ consumer.cancel();
+ consumerThread.sync();
+
+ assertEquals(numShards, testShardStates.size());
+
+ for (int i = 0; i < numShards; i++) {
+ fetcher.emitRecordAndUpdateState("record-" + i, 10L, i, new SequenceNumber("seq-num-1"));
+ assertEquals(new SequenceNumber("seq-num-1"), testShardStates.get(i).getLastProcessedSequenceNum());
+ assertEquals(new StreamRecord<>("record-" + i, 10L), sourceContext.removeLatestOutput());
--- End diff --
indentation
---
[GitHub] flink pull request #5269: [FLINK-6004] [kinesis] Allow FlinkKinesisConsumer ...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/5269#discussion_r165039352
--- Diff: flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/KinesisDataFetcher.java ---
@@ -484,7 +484,10 @@ protected Properties getConsumerConfiguration() {
*/
protected final void emitRecordAndUpdateState(T record, long recordTimestamp, int shardStateIndex, SequenceNumber lastSequenceNumber) {
synchronized (checkpointLock) {
- sourceContext.collectWithTimestamp(record, recordTimestamp);
+ if (record != null) {
+ sourceContext.collectWithTimestamp(record, recordTimestamp);
--- End diff --
We currently do not have a log for that.
I'll add a warning log if the record is null.
---
[GitHub] flink issue #5269: [FLINK-6004] [kinesis] Allow FlinkKinesisConsumer to skip...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on the issue:
https://github.com/apache/flink/pull/5269
@zentol
The contract is that if `KinesisDeserializationSchema.deserialize()` returns `null`, that record will be skipped. I agree it makes sense to have a add a warning log if this happens.
---
[GitHub] flink pull request #5269: [FLINK-6004] [kinesis] Allow FlinkKinesisConsumer ...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/5269
---
[GitHub] flink issue #5269: [FLINK-6004] [kinesis] Allow FlinkKinesisConsumer to skip...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on the issue:
https://github.com/apache/flink/pull/5269
Thanks, merging ..
---
[GitHub] flink issue #5269: [FLINK-6004] [kinesis] Allow FlinkKinesisConsumer to skip...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the issue:
https://github.com/apache/flink/pull/5269
looks good, +1
---
[GitHub] flink pull request #5269: [FLINK-6004] [kinesis] Allow FlinkKinesisConsumer ...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/5269#discussion_r165038908
--- Diff: flink-connectors/flink-connector-kinesis/src/test/java/org/apache/flink/streaming/connectors/kinesis/internals/KinesisDataFetcherTest.java ---
@@ -86,6 +88,56 @@ public void testIfNoShardsAreFoundShouldThrowException() throws Exception {
fetcher.runFetcher(); // this should throw RuntimeException
}
+ @Test
+ public void testSkipCorruptedRecord() throws Exception {
+ final String stream = "fakeStream";
+ final int numShards = 3;
+
+ final LinkedList<KinesisStreamShardState> testShardStates = new LinkedList<>();
+ final TestSourceContext<String> sourceContext = new TestSourceContext<>();
+
+ final TestableKinesisDataFetcher<String> fetcher = new TestableKinesisDataFetcher<>(
+ Collections.singletonList(stream),
+ sourceContext,
+ TestUtils.getStandardProperties(),
+ new KinesisDeserializationSchemaWrapper<>(new SimpleStringSchema()),
+ 1,
+ 0,
+ new AtomicReference<>(),
+ testShardStates,
+ new HashMap<>(),
+ FakeKinesisBehavioursFactory.nonReshardedStreamsBehaviour(Collections.singletonMap(stream, numShards)));
+
+ // FlinkKinesisConsumer is responsible for setting up the fetcher before it can be run;
+ // run the consumer until it reaches the point where the fetcher starts to run
+ final DummyFlinkKafkaConsumer<String> consumer = new DummyFlinkKafkaConsumer<>(TestUtils.getStandardProperties(), fetcher, 1, 0);
+
+ CheckedThread consumerThread = new CheckedThread() {
+ @Override
+ public void go() throws Exception {
+ consumer.run(new TestSourceContext<>());
+ }
+ };
+ consumerThread.start();
+
+ fetcher.waitUntilRun();
+ consumer.cancel();
+ consumerThread.sync();
+
+ assertEquals(numShards, testShardStates.size());
+
+ for (int i = 0; i < numShards; i++) {
+ fetcher.emitRecordAndUpdateState("record-" + i, 10L, i, new SequenceNumber("seq-num-1"));
+ assertEquals(new SequenceNumber("seq-num-1"), testShardStates.get(i).getLastProcessedSequenceNum());
+ assertEquals(new StreamRecord<>("record-" + i, 10L), sourceContext.removeLatestOutput());
--- End diff --
Will fix.
---
[GitHub] flink pull request #5269: [FLINK-6004] [kinesis] Allow FlinkKinesisConsumer ...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/5269#discussion_r165019903
--- Diff: flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/KinesisDataFetcher.java ---
@@ -484,7 +484,10 @@ protected Properties getConsumerConfiguration() {
*/
protected final void emitRecordAndUpdateState(T record, long recordTimestamp, int shardStateIndex, SequenceNumber lastSequenceNumber) {
synchronized (checkpointLock) {
- sourceContext.collectWithTimestamp(record, recordTimestamp);
+ if (record != null) {
+ sourceContext.collectWithTimestamp(record, recordTimestamp);
--- End diff --
Are we silently skipping the record or do we log _somewhere_ that a record was invalid?
---
[GitHub] flink issue #5269: [FLINK-6004] [kinesis] Allow FlinkKinesisConsumer to skip...
Posted by tzulitai <gi...@git.apache.org>.
Github user tzulitai commented on the issue:
https://github.com/apache/flink/pull/5269
@zentol I've rebased, and adding a warning when skipping records. Could you have another quick look?
---