Use KafkaSource instead of FlinkKafkaConsumer

[ardakuyumcu]

This pull request includes significant updates to the runners/flink module, primarily focusing on upgrading dependencies and refactoring the Kafka-related code to use updated classes and methods. The most important changes include updating the dependencies in flink_runner.gradle, refactoring the LyftFlinkStreamingPortableTranslations class to use the LyftKafkaSourceBuilder and KafkaSource, and modifying the ByteArrayWindowedValueSchema class to implement KafkaRecordDeserializationSchema.

[kjbilton]

@ardakuyumcu i'm seeing errors when deploying a pipeline using this change, and i noticed this log -- any idea how it could be related?

[ardakuyumcu]

@kjbilton I'm relatively new to the flink/beam ecosystem, but seems like perhaps the bundled Kafka client version is too old?

Our Flink version is requesting Kafka 3.2.3 here https://github.com/lyft/flink/blob/709df019f5c3a5d7945c5931d9aa3c0c70560c35/flink-connectors/flink-connector-kafka/pom.xml#L38

but might be Beam is building with an older version of Kafka? the only reference I could find (not sure if related) is to 2.4.1 here

beam/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy

Line 553 in a7a0715

def kafka_version = "2.4.1"

which could explain why you're seeing NoSuchMethodError

@maghamravi any ideas?

@kjbilton would you be able to see what version of the kafka java client your container is running with?

[kjbilton]

@kjbilton I'm relatively new to the flink/beam ecosystem, but seems like perhaps the bundled Kafka client version is too old?

Our Flink version is requesting Kafka 3.2.3 here https://github.com/lyft/flink/blob/709df019f5c3a5d7945c5931d9aa3c0c70560c35/flink-connectors/flink-connector-kafka/pom.xml#L38

but might be Beam is building with an older version of Kafka? the only reference I could find (not sure if related) is to 2.4.1 here

beam/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy

Line 553 in a7a0715

def kafka_version = "2.4.1"

which could explain why you're seeing NoSuchMethodError
@maghamravi any ideas?

@kjbilton would you be able to see what version of the kafka java client your container is running with?

Ah, good catch. Yes, the logs indicate that we are using 2.4.1. I see that other apps using this source are running 3.1.0 [logs]. I'll take a look at upgrading Kafka here.

[kjbilton]

Upgrading to 3.1.0 appears to be running fine -- going to take a closer look to validate, but so far it all looks good! Tested here: #78