gadhiyakrutik
06/12/2023, 11:08 AMMartijn Visser
06/12/2023, 11:31 AMgadhiyakrutik
06/12/2023, 11:53 AMpublic 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 occuredgadhiyakrutik
06/12/2023, 11:55 AM