@Test @SuppressWarnings("unchecked") public void testFetcherShouldNotBeRestoringFromFailureIfNotRestoringFromCheckpoint() throws Exception { KinesisDataFetcher mockedFetcher = mockKinesisDataFetcher(); PowerMockito.mockStatic(KinesisConfigUtil.class); PowerMockito.doNothing().when(KinesisConfigUtil.class); TestableFlinkKinesisConsumer consumer = new TestableFlinkKinesisConsumer( "fakeStream", new Properties(), 10, 2); consumer.open(new Configuration()); consumer.run(Mockito.mock(SourceFunction.SourceContext.class)); }
@Override public void run(SourceContext<T> sourceContext) throws Exception { KinesisDataFetcher<T> fetcher = createFetcher(streams, sourceContext, getRuntimeContext(), configProps, deserializer); List<StreamShardHandle> allShards = fetcher.discoverNewShardsToSubscribe(); for (StreamShardHandle shard : allShards) { StreamShardMetadata.EquivalenceWrapper kinesisStreamShard = new StreamShardMetadata.EquivalenceWrapper(KinesisDataFetcher.convertToStreamShardMetadata(shard)); if (sequenceNumsToRestore != null) { if (sequenceNumsToRestore.containsKey(kinesisStreamShard)) { fetcher.registerNewSubscribedShardState( new KinesisStreamShardState(kinesisStreamShard.getShardMetadata(), shard, sequenceNumsToRestore.get(kinesisStreamShard))); if (LOG.isInfoEnabled()) { LOG.info("Subtask {} is seeding the fetcher with restored shard {}," + " starting state set to the restored sequence number {}", getRuntimeContext().getIndexOfThisSubtask(), shard.toString(), sequenceNumsToRestore.get(kinesisStreamShard)); } } else { fetcher.registerNewSubscribedShardState( new KinesisStreamShardState(kinesisStreamShard.getShardMetadata(), shard, SentinelSequenceNumber.SENTINEL_EARLIEST_SEQUENCE_NUM.get())); if (LOG.isInfoEnabled()) { LOG.info("Subtask {} is seeding the fetcher with new discovered shard {}," + " starting state set to the SENTINEL_EARLIEST_SEQUENCE_NUM", getRuntimeContext().getIndexOfThisSubtask(), shard.toString()); } } } else { SentinelSequenceNumber startingSeqNum = InitialPosition.valueOf(configProps.getProperty( ConsumerConfigConstants.STREAM_INITIAL_POSITION, ConsumerConfigConstants.DEFAULT_STREAM_INITIAL_POSITION)).toSentinelSequenceNumber(); fetcher.registerNewSubscribedShardState( new KinesisStreamShardState(kinesisStreamShard.getShardMetadata(), shard, startingSeqNum.get())); if (LOG.isInfoEnabled()) { LOG.info("Subtask {} will be seeded with initial shard {}, starting state set as sequence number {}", getRuntimeContext().getIndexOfThisSubtask(), shard.toString(), startingSeqNum.get()); } } } if (!running) { return; } this.fetcher = fetcher; fetcher.runFetcher(); fetcher.awaitTermination(); sourceContext.close(); }
FlinkKinesisConsumer extends RichParallelSourceFunction<T> implements
		ResultTypeQueryable<T>,
		CheckpointedFunction { @Override public void run(SourceContext<T> sourceContext) throws Exception { KinesisDataFetcher<T> fetcher = createFetcher(streams, sourceContext, getRuntimeContext(), configProps, deserializer); List<StreamShardHandle> allShards = fetcher.discoverNewShardsToSubscribe(); for (StreamShardHandle shard : allShards) { StreamShardMetadata.EquivalenceWrapper kinesisStreamShard = new StreamShardMetadata.EquivalenceWrapper(KinesisDataFetcher.convertToStreamShardMetadata(shard)); if (sequenceNumsToRestore != null) { if (sequenceNumsToRestore.containsKey(kinesisStreamShard)) { fetcher.registerNewSubscribedShardState( new KinesisStreamShardState(kinesisStreamShard.getShardMetadata(), shard, sequenceNumsToRestore.get(kinesisStreamShard))); if (LOG.isInfoEnabled()) { LOG.info("Subtask {} is seeding the fetcher with restored shard {}," + " starting state set to the restored sequence number {}", getRuntimeContext().getIndexOfThisSubtask(), shard.toString(), sequenceNumsToRestore.get(kinesisStreamShard)); } } else { fetcher.registerNewSubscribedShardState( new KinesisStreamShardState(kinesisStreamShard.getShardMetadata(), shard, SentinelSequenceNumber.SENTINEL_EARLIEST_SEQUENCE_NUM.get())); if (LOG.isInfoEnabled()) { LOG.info("Subtask {} is seeding the fetcher with new discovered shard {}," + " starting state set to the SENTINEL_EARLIEST_SEQUENCE_NUM", getRuntimeContext().getIndexOfThisSubtask(), shard.toString()); } } } else { SentinelSequenceNumber startingSeqNum = InitialPosition.valueOf(configProps.getProperty( ConsumerConfigConstants.STREAM_INITIAL_POSITION, ConsumerConfigConstants.DEFAULT_STREAM_INITIAL_POSITION)).toSentinelSequenceNumber(); fetcher.registerNewSubscribedShardState( new KinesisStreamShardState(kinesisStreamShard.getShardMetadata(), shard, startingSeqNum.get())); if (LOG.isInfoEnabled()) { LOG.info("Subtask {} will be seeded with initial shard {}, starting state set as sequence number {}", getRuntimeContext().getIndexOfThisSubtask(), shard.toString(), startingSeqNum.get()); } } } if (!running) { return; } this.fetcher = fetcher; fetcher.runFetcher(); fetcher.awaitTermination(); sourceContext.close(); } }
FlinkKinesisConsumer extends RichParallelSourceFunction<T> implements
		ResultTypeQueryable<T>,
		CheckpointedFunction { @Override public void run(SourceContext<T> sourceContext) throws Exception { KinesisDataFetcher<T> fetcher = createFetcher(streams, sourceContext, getRuntimeContext(), configProps, deserializer); List<StreamShardHandle> allShards = fetcher.discoverNewShardsToSubscribe(); for (StreamShardHandle shard : allShards) { StreamShardMetadata.EquivalenceWrapper kinesisStreamShard = new StreamShardMetadata.EquivalenceWrapper(KinesisDataFetcher.convertToStreamShardMetadata(shard)); if (sequenceNumsToRestore != null) { if (sequenceNumsToRestore.containsKey(kinesisStreamShard)) { fetcher.registerNewSubscribedShardState( new KinesisStreamShardState(kinesisStreamShard.getShardMetadata(), shard, sequenceNumsToRestore.get(kinesisStreamShard))); if (LOG.isInfoEnabled()) { LOG.info("Subtask {} is seeding the fetcher with restored shard {}," + " starting state set to the restored sequence number {}", getRuntimeContext().getIndexOfThisSubtask(), shard.toString(), sequenceNumsToRestore.get(kinesisStreamShard)); } } else { fetcher.registerNewSubscribedShardState( new KinesisStreamShardState(kinesisStreamShard.getShardMetadata(), shard, SentinelSequenceNumber.SENTINEL_EARLIEST_SEQUENCE_NUM.get())); if (LOG.isInfoEnabled()) { LOG.info("Subtask {} is seeding the fetcher with new discovered shard {}," + " starting state set to the SENTINEL_EARLIEST_SEQUENCE_NUM", getRuntimeContext().getIndexOfThisSubtask(), shard.toString()); } } } else { SentinelSequenceNumber startingSeqNum = InitialPosition.valueOf(configProps.getProperty( ConsumerConfigConstants.STREAM_INITIAL_POSITION, ConsumerConfigConstants.DEFAULT_STREAM_INITIAL_POSITION)).toSentinelSequenceNumber(); fetcher.registerNewSubscribedShardState( new KinesisStreamShardState(kinesisStreamShard.getShardMetadata(), shard, startingSeqNum.get())); if (LOG.isInfoEnabled()) { LOG.info("Subtask {} will be seeded with initial shard {}, starting state set as sequence number {}", getRuntimeContext().getIndexOfThisSubtask(), shard.toString(), startingSeqNum.get()); } } } if (!running) { return; } this.fetcher = fetcher; fetcher.runFetcher(); fetcher.awaitTermination(); sourceContext.close(); } FlinkKinesisConsumer(String stream, DeserializationSchema<T> deserializer, Properties configProps);  FlinkKinesisConsumer(String stream, KinesisDeserializationSchema<T> deserializer, Properties configProps);  FlinkKinesisConsumer(List<String> streams, KinesisDeserializationSchema<T> deserializer, Properties configProps); }
FlinkKinesisConsumer extends RichParallelSourceFunction<T> implements
		ResultTypeQueryable<T>,
		CheckpointedFunction { @Override public void run(SourceContext<T> sourceContext) throws Exception { KinesisDataFetcher<T> fetcher = createFetcher(streams, sourceContext, getRuntimeContext(), configProps, deserializer); List<StreamShardHandle> allShards = fetcher.discoverNewShardsToSubscribe(); for (StreamShardHandle shard : allShards) { StreamShardMetadata.EquivalenceWrapper kinesisStreamShard = new StreamShardMetadata.EquivalenceWrapper(KinesisDataFetcher.convertToStreamShardMetadata(shard)); if (sequenceNumsToRestore != null) { if (sequenceNumsToRestore.containsKey(kinesisStreamShard)) { fetcher.registerNewSubscribedShardState( new KinesisStreamShardState(kinesisStreamShard.getShardMetadata(), shard, sequenceNumsToRestore.get(kinesisStreamShard))); if (LOG.isInfoEnabled()) { LOG.info("Subtask {} is seeding the fetcher with restored shard {}," + " starting state set to the restored sequence number {}", getRuntimeContext().getIndexOfThisSubtask(), shard.toString(), sequenceNumsToRestore.get(kinesisStreamShard)); } } else { fetcher.registerNewSubscribedShardState( new KinesisStreamShardState(kinesisStreamShard.getShardMetadata(), shard, SentinelSequenceNumber.SENTINEL_EARLIEST_SEQUENCE_NUM.get())); if (LOG.isInfoEnabled()) { LOG.info("Subtask {} is seeding the fetcher with new discovered shard {}," + " starting state set to the SENTINEL_EARLIEST_SEQUENCE_NUM", getRuntimeContext().getIndexOfThisSubtask(), shard.toString()); } } } else { SentinelSequenceNumber startingSeqNum = InitialPosition.valueOf(configProps.getProperty( ConsumerConfigConstants.STREAM_INITIAL_POSITION, ConsumerConfigConstants.DEFAULT_STREAM_INITIAL_POSITION)).toSentinelSequenceNumber(); fetcher.registerNewSubscribedShardState( new KinesisStreamShardState(kinesisStreamShard.getShardMetadata(), shard, startingSeqNum.get())); if (LOG.isInfoEnabled()) { LOG.info("Subtask {} will be seeded with initial shard {}, starting state set as sequence number {}", getRuntimeContext().getIndexOfThisSubtask(), shard.toString(), startingSeqNum.get()); } } } if (!running) { return; } this.fetcher = fetcher; fetcher.runFetcher(); fetcher.awaitTermination(); sourceContext.close(); } FlinkKinesisConsumer(String stream, DeserializationSchema<T> deserializer, Properties configProps);  FlinkKinesisConsumer(String stream, KinesisDeserializationSchema<T> deserializer, Properties configProps);  FlinkKinesisConsumer(List<String> streams, KinesisDeserializationSchema<T> deserializer, Properties configProps); KinesisShardAssigner getShardAssigner(); void setShardAssigner(KinesisShardAssigner shardAssigner); AssignerWithPeriodicWatermarks<T> getPeriodicWatermarkAssigner(); void setPeriodicWatermarkAssigner(
		AssignerWithPeriodicWatermarks<T> periodicWatermarkAssigner); @Override void run(SourceContext<T> sourceContext); @Override void cancel(); @Override void close(); @Override TypeInformation<T> getProducedType(); @Override void initializeState(FunctionInitializationContext context); @Override void snapshotState(FunctionSnapshotContext context); }
FlinkKinesisConsumer extends RichParallelSourceFunction<T> implements
		ResultTypeQueryable<T>,
		CheckpointedFunction { @Override public void run(SourceContext<T> sourceContext) throws Exception { KinesisDataFetcher<T> fetcher = createFetcher(streams, sourceContext, getRuntimeContext(), configProps, deserializer); List<StreamShardHandle> allShards = fetcher.discoverNewShardsToSubscribe(); for (StreamShardHandle shard : allShards) { StreamShardMetadata.EquivalenceWrapper kinesisStreamShard = new StreamShardMetadata.EquivalenceWrapper(KinesisDataFetcher.convertToStreamShardMetadata(shard)); if (sequenceNumsToRestore != null) { if (sequenceNumsToRestore.containsKey(kinesisStreamShard)) { fetcher.registerNewSubscribedShardState( new KinesisStreamShardState(kinesisStreamShard.getShardMetadata(), shard, sequenceNumsToRestore.get(kinesisStreamShard))); if (LOG.isInfoEnabled()) { LOG.info("Subtask {} is seeding the fetcher with restored shard {}," + " starting state set to the restored sequence number {}", getRuntimeContext().getIndexOfThisSubtask(), shard.toString(), sequenceNumsToRestore.get(kinesisStreamShard)); } } else { fetcher.registerNewSubscribedShardState( new KinesisStreamShardState(kinesisStreamShard.getShardMetadata(), shard, SentinelSequenceNumber.SENTINEL_EARLIEST_SEQUENCE_NUM.get())); if (LOG.isInfoEnabled()) { LOG.info("Subtask {} is seeding the fetcher with new discovered shard {}," + " starting state set to the SENTINEL_EARLIEST_SEQUENCE_NUM", getRuntimeContext().getIndexOfThisSubtask(), shard.toString()); } } } else { SentinelSequenceNumber startingSeqNum = InitialPosition.valueOf(configProps.getProperty( ConsumerConfigConstants.STREAM_INITIAL_POSITION, ConsumerConfigConstants.DEFAULT_STREAM_INITIAL_POSITION)).toSentinelSequenceNumber(); fetcher.registerNewSubscribedShardState( new KinesisStreamShardState(kinesisStreamShard.getShardMetadata(), shard, startingSeqNum.get())); if (LOG.isInfoEnabled()) { LOG.info("Subtask {} will be seeded with initial shard {}, starting state set as sequence number {}", getRuntimeContext().getIndexOfThisSubtask(), shard.toString(), startingSeqNum.get()); } } } if (!running) { return; } this.fetcher = fetcher; fetcher.runFetcher(); fetcher.awaitTermination(); sourceContext.close(); } FlinkKinesisConsumer(String stream, DeserializationSchema<T> deserializer, Properties configProps);  FlinkKinesisConsumer(String stream, KinesisDeserializationSchema<T> deserializer, Properties configProps);  FlinkKinesisConsumer(List<String> streams, KinesisDeserializationSchema<T> deserializer, Properties configProps); KinesisShardAssigner getShardAssigner(); void setShardAssigner(KinesisShardAssigner shardAssigner); AssignerWithPeriodicWatermarks<T> getPeriodicWatermarkAssigner(); void setPeriodicWatermarkAssigner(
		AssignerWithPeriodicWatermarks<T> periodicWatermarkAssigner); @Override void run(SourceContext<T> sourceContext); @Override void cancel(); @Override void close(); @Override TypeInformation<T> getProducedType(); @Override void initializeState(FunctionInitializationContext context); @Override void snapshotState(FunctionSnapshotContext context);  }