https://flink.apache.org/ logo
Join Slack
Powered by
```env = StreamExecutionEnvironment.get_execution_...
# random
a
Copy code
env = StreamExecutionEnvironment.get_execution_environment()
table_env = StreamTableEnvironment.create(env)
env.add_jars("file:///home/muramsettisai/miniconda3/envs/new_flink_env/lib/python3.10/site-packages/pyflink/flink-sql-connector-kafka-3.1.0-1.18.jar")  # Replace with your actual path


kafka_source1 = KafkaSource.builder() \
    .set_bootstrap_servers("-----") \
    .set_topics("----") \
    .set_group_id('----')\
    .set_value_only_deserializer(SimpleStringSchema()).set_starting_offsets(KafkaOffsetsInitializer.committed_offsets(KafkaOffsetResetStrategy.LATEST)) \
    .build()
data_stream1 = env.from_source(  
    source=kafka_source1,
    watermark_strategy=WatermarkStrategy.no_watermarks(),
    source_name="Kafka Source 1",
)
j
Your group Id '----' is running late , because you might have started this program hours before with the same groupId and now when we starting the same program later with same group id you will only get Kafka data from where you left ( 1 or 2 hour ago )
a
oh thanks
how to rectify that error in production? do we have to change group id frequently?
but even i change group id to some random group id iam getting old data some times
p
You may want to consume data from 
latest
instead of 
earliest
a
Copy code
i mentioned latest in .set_value_only_deserializer(SimpleStringSchema()).set_starting_offsets(KafkaOffsetsInitializer.committed_offsets(KafkaOffsetResetStrategy.LATEST))
Open in Slack
PreviousNext
Powered by