Hi am using this code to read and write from kafka but not able to find group_id=flinktestpers but I was able to see when I was using FlinkKafkaConsumer instead of kafkaSource

from pyflink.common.serialization import SimpleStringSchema
from pyflink.common.typeinfo import Types
from pyflink.common.watermark_strategy import WatermarkStrategy
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors.kafka import KafkaSource, KafkaSink, KafkaOffsetsInitializer, KafkaRecordSerializationSchema, DeliveryGuarantee

def kafkaread():
    env = StreamExecutionEnvironment.get_execution_environment()

    brokers = 'cluster-0-kafka-bootstrap.strimzi.svc.cluster.local:9092'
    source = KafkaSource.builder()\
        .set_bootstrap_servers(brokers) \
        .set_topics("locations") \
        .set_group_id("flinktestpers") \
        .set_starting_offsets(KafkaOffsetsInitializer.earliest()) \
        .set_value_only_deserializer(SimpleStringSchema()) \
        .build()

    ds = env.from_source(source, WatermarkStrategy.no_watermarks(), "Kafka Source")
    ds.print()

    # Define the Kafka sink
    sink = KafkaSink.builder() \
        .set_bootstrap_servers(brokers) \
        .set_record_serializer(
            KafkaRecordSerializationSchema.builder()
                .set_topic("flinksink") # Set your output topic here
                .set_value_serialization_schema(SimpleStringSchema())
                .build()
        ) \
        .set_delivery_guarantee(DeliveryGuarantee.AT_LEAST_ONCE) \
        .build()

    ds.sink_to(sink)

    env.execute('kafkaread')

if __name__ == '__main__':
    kafkaread()

FLINK_VERSION=1.16.2

@kapil singh Could you check if the job is really running? You could check it via the Flink Web UI. You could also check if there are error logs in the logging file.

yes job was running ....and no error in error logs