if failure occurred at some point and lets say Fli...
# random
g
if failure occurred at some point and lets say Flink took 1 min to recover from that than what would happen to data streamed within that 1 min, does Flink keep track of how much is data streamed?
m
It depends if your connectors have fault-tolerance (at least once, exactly once etc) support and if you have enabled checkpointing. If that's the case, Flink would just resume with reading where it stopped before the failure
g
Copy code
public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.enableCheckpointing(10000, CheckpointingMode.EXACTLY_ONCE);

        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000);

        env.getCheckpointConfig().setExternalizedCheckpointCleanup(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        EmbeddedRocksDBStateBackend stateBackend = new EmbeddedRocksDBStateBackend(true);
        stateBackend.setPredefinedOptions(PredefinedOptions.FLASH_SSD_OPTIMIZED);
        env.setStateBackend(stateBackend);

        env.getCheckpointConfig().setCheckpointStorage("<hdfs://127.0.0.1:9000/flink/checkpoints>");

        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
                5,
                org.apache.flink.api.common.time.Time.of(10, TimeUnit.SECONDS))
        );

        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("127.0.0.1:9092")
                .setTopics("status")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .setBounded(OffsetsInitializer.latest()) // tells flink to stop at latest offset
                .build();
}
I have this much setup done, so this would recover if any failure occured
could you suggest anything that would be nice to have and i am missing with