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
[FLINK-6004] [kinesis] Allow FlinkKinesisConsumer to skip non-deserializable records #5269
Conversation
@@ -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); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Are we silently skipping the record or do we log somewhere that a record was invalid?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We currently do not have a log for that.
I'll add a warning log if the record is null.
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()); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
indentation
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Will fix.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Changes look good but I'm concerned about the visibility of this behavior.
@zentol |
…nesisDataFetcherTest 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.
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.
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.
@zentol I've rebased, and added a warning when skipping records. Could you have another quick look? |
looks good, +1 |
Thanks, merging .. |
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. This closes apache#5269.
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. This closes apache#5269.
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
KinesisDeserializationSchema
thatnull
can be returned if a message cannot be deserialized.record
isnull
inKinesisDataFetcher::emitRecordAndUpdateState(...)
, do not collect any output for the record.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:
@Public(Evolving)
: (yes / no)Documentation