I used KafkaSource to consume the messages from Kafka. I found only new messages were received while the old existing message not. I tried to use a new consumer group and update the parameter "auto.offset.reset = latest" to "earliest", but this does not work.

tier2.sources.source1.kafka.consumer.group.id = test-consumer-group-new
tier2.sources.source1.kafka.consumer.auto.offset.reset = earliest

Anyone knows how to make KafkaSource consume the existing messages?
Thanks a lot for any advice!