@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))); final DummyFlinkKinesisConsumer<String> consumer = new DummyFlinkKinesisConsumer<>(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()); } fetcher.emitRecordAndUpdateState(null, 10L, 1, new SequenceNumber("seq-num-2")); assertEquals(new SequenceNumber("seq-num-2"), testShardStates.get(1).getLastProcessedSequenceNum()); assertEquals(null, sourceContext.removeLatestOutput()); }
protected void emitRecordAndUpdateState(T record, long recordTimestamp, int shardStateIndex, SequenceNumber lastSequenceNumber) { if (periodicWatermarkAssigner != null) { ShardWatermarkState sws = shardWatermarks.get(shardStateIndex); Preconditions.checkNotNull( sws, "shard watermark state initialized in registerNewSubscribedShardState"); recordTimestamp = sws.periodicWatermarkAssigner.extractTimestamp(record, sws.lastRecordTimestamp); sws.lastRecordTimestamp = recordTimestamp; sws.lastUpdated = getCurrentTimeMillis(); } synchronized (checkpointLock) { if (record != null) { sourceContext.collectWithTimestamp(record, recordTimestamp); } else { LOG.warn("Skipping non-deserializable record at sequence number {} of shard {}.", lastSequenceNumber, subscribedShardsState.get(shardStateIndex).getStreamShardHandle()); } updateState(shardStateIndex, lastSequenceNumber); } }
KinesisDataFetcher { protected void emitRecordAndUpdateState(T record, long recordTimestamp, int shardStateIndex, SequenceNumber lastSequenceNumber) { if (periodicWatermarkAssigner != null) { ShardWatermarkState sws = shardWatermarks.get(shardStateIndex); Preconditions.checkNotNull( sws, "shard watermark state initialized in registerNewSubscribedShardState"); recordTimestamp = sws.periodicWatermarkAssigner.extractTimestamp(record, sws.lastRecordTimestamp); sws.lastRecordTimestamp = recordTimestamp; sws.lastUpdated = getCurrentTimeMillis(); } synchronized (checkpointLock) { if (record != null) { sourceContext.collectWithTimestamp(record, recordTimestamp); } else { LOG.warn("Skipping non-deserializable record at sequence number {} of shard {}.", lastSequenceNumber, subscribedShardsState.get(shardStateIndex).getStreamShardHandle()); } updateState(shardStateIndex, lastSequenceNumber); } } }
KinesisDataFetcher { protected void emitRecordAndUpdateState(T record, long recordTimestamp, int shardStateIndex, SequenceNumber lastSequenceNumber) { if (periodicWatermarkAssigner != null) { ShardWatermarkState sws = shardWatermarks.get(shardStateIndex); Preconditions.checkNotNull( sws, "shard watermark state initialized in registerNewSubscribedShardState"); recordTimestamp = sws.periodicWatermarkAssigner.extractTimestamp(record, sws.lastRecordTimestamp); sws.lastRecordTimestamp = recordTimestamp; sws.lastUpdated = getCurrentTimeMillis(); } synchronized (checkpointLock) { if (record != null) { sourceContext.collectWithTimestamp(record, recordTimestamp); } else { LOG.warn("Skipping non-deserializable record at sequence number {} of shard {}.", lastSequenceNumber, subscribedShardsState.get(shardStateIndex).getStreamShardHandle()); } updateState(shardStateIndex, lastSequenceNumber); } } KinesisDataFetcher(List<String> streams,
							SourceFunction.SourceContext<T> sourceContext,
							RuntimeContext runtimeContext,
							Properties configProps,
							KinesisDeserializationSchema<T> deserializationSchema,
							KinesisShardAssigner shardAssigner,
							AssignerWithPeriodicWatermarks<T> periodicWatermarkAssigner); @VisibleForTesting protected  KinesisDataFetcher(List<String> streams,
								SourceFunction.SourceContext<T> sourceContext,
								Object checkpointLock,
								RuntimeContext runtimeContext,
								Properties configProps,
								KinesisDeserializationSchema<T> deserializationSchema,
								KinesisShardAssigner shardAssigner,
								AssignerWithPeriodicWatermarks<T> periodicWatermarkAssigner,
								AtomicReference<Throwable> error,
								List<KinesisStreamShardState> subscribedShardsState,
								HashMap<String, String> subscribedStreamsToLastDiscoveredShardIds,
								FlinkKinesisProxyFactory kinesisProxyFactory); }
KinesisDataFetcher { protected void emitRecordAndUpdateState(T record, long recordTimestamp, int shardStateIndex, SequenceNumber lastSequenceNumber) { if (periodicWatermarkAssigner != null) { ShardWatermarkState sws = shardWatermarks.get(shardStateIndex); Preconditions.checkNotNull( sws, "shard watermark state initialized in registerNewSubscribedShardState"); recordTimestamp = sws.periodicWatermarkAssigner.extractTimestamp(record, sws.lastRecordTimestamp); sws.lastRecordTimestamp = recordTimestamp; sws.lastUpdated = getCurrentTimeMillis(); } synchronized (checkpointLock) { if (record != null) { sourceContext.collectWithTimestamp(record, recordTimestamp); } else { LOG.warn("Skipping non-deserializable record at sequence number {} of shard {}.", lastSequenceNumber, subscribedShardsState.get(shardStateIndex).getStreamShardHandle()); } updateState(shardStateIndex, lastSequenceNumber); } } KinesisDataFetcher(List<String> streams,
							SourceFunction.SourceContext<T> sourceContext,
							RuntimeContext runtimeContext,
							Properties configProps,
							KinesisDeserializationSchema<T> deserializationSchema,
							KinesisShardAssigner shardAssigner,
							AssignerWithPeriodicWatermarks<T> periodicWatermarkAssigner); @VisibleForTesting protected  KinesisDataFetcher(List<String> streams,
								SourceFunction.SourceContext<T> sourceContext,
								Object checkpointLock,
								RuntimeContext runtimeContext,
								Properties configProps,
								KinesisDeserializationSchema<T> deserializationSchema,
								KinesisShardAssigner shardAssigner,
								AssignerWithPeriodicWatermarks<T> periodicWatermarkAssigner,
								AtomicReference<Throwable> error,
								List<KinesisStreamShardState> subscribedShardsState,
								HashMap<String, String> subscribedStreamsToLastDiscoveredShardIds,
								FlinkKinesisProxyFactory kinesisProxyFactory); void runFetcher(); HashMap<StreamShardMetadata, SequenceNumber> snapshotState(); void shutdownFetcher(); void awaitTermination(); void advanceLastDiscoveredShardOfStream(String stream, String shardId); List<StreamShardHandle> discoverNewShardsToSubscribe(); int registerNewSubscribedShardState(KinesisStreamShardState newSubscribedShardState); static boolean isThisSubtaskShouldSubscribeTo(int shardHash,
														int totalNumberOfConsumerSubtasks,
														int indexOfThisConsumerSubtask); @VisibleForTesting List<KinesisStreamShardState> getSubscribedShardsState(); static StreamShardMetadata convertToStreamShardMetadata(StreamShardHandle streamShardHandle); static StreamShardHandle convertToStreamShardHandle(StreamShardMetadata streamShardMetadata); }
KinesisDataFetcher { protected void emitRecordAndUpdateState(T record, long recordTimestamp, int shardStateIndex, SequenceNumber lastSequenceNumber) { if (periodicWatermarkAssigner != null) { ShardWatermarkState sws = shardWatermarks.get(shardStateIndex); Preconditions.checkNotNull( sws, "shard watermark state initialized in registerNewSubscribedShardState"); recordTimestamp = sws.periodicWatermarkAssigner.extractTimestamp(record, sws.lastRecordTimestamp); sws.lastRecordTimestamp = recordTimestamp; sws.lastUpdated = getCurrentTimeMillis(); } synchronized (checkpointLock) { if (record != null) { sourceContext.collectWithTimestamp(record, recordTimestamp); } else { LOG.warn("Skipping non-deserializable record at sequence number {} of shard {}.", lastSequenceNumber, subscribedShardsState.get(shardStateIndex).getStreamShardHandle()); } updateState(shardStateIndex, lastSequenceNumber); } } KinesisDataFetcher(List<String> streams,
							SourceFunction.SourceContext<T> sourceContext,
							RuntimeContext runtimeContext,
							Properties configProps,
							KinesisDeserializationSchema<T> deserializationSchema,
							KinesisShardAssigner shardAssigner,
							AssignerWithPeriodicWatermarks<T> periodicWatermarkAssigner); @VisibleForTesting protected  KinesisDataFetcher(List<String> streams,
								SourceFunction.SourceContext<T> sourceContext,
								Object checkpointLock,
								RuntimeContext runtimeContext,
								Properties configProps,
								KinesisDeserializationSchema<T> deserializationSchema,
								KinesisShardAssigner shardAssigner,
								AssignerWithPeriodicWatermarks<T> periodicWatermarkAssigner,
								AtomicReference<Throwable> error,
								List<KinesisStreamShardState> subscribedShardsState,
								HashMap<String, String> subscribedStreamsToLastDiscoveredShardIds,
								FlinkKinesisProxyFactory kinesisProxyFactory); void runFetcher(); HashMap<StreamShardMetadata, SequenceNumber> snapshotState(); void shutdownFetcher(); void awaitTermination(); void advanceLastDiscoveredShardOfStream(String stream, String shardId); List<StreamShardHandle> discoverNewShardsToSubscribe(); int registerNewSubscribedShardState(KinesisStreamShardState newSubscribedShardState); static boolean isThisSubtaskShouldSubscribeTo(int shardHash,
														int totalNumberOfConsumerSubtasks,
														int indexOfThisConsumerSubtask); @VisibleForTesting List<KinesisStreamShardState> getSubscribedShardsState(); static StreamShardMetadata convertToStreamShardMetadata(StreamShardHandle streamShardHandle); static StreamShardHandle convertToStreamShardHandle(StreamShardMetadata streamShardMetadata); static final KinesisShardAssigner DEFAULT_SHARD_ASSIGNER; }