Hi Matthias, Thank you for your reply!
There should not be a dependency for 3.0.x in my docker image, I only add 3.2.0 explicitly. When connecting to the running container I also can’t find any reference to 3.0.x. I reverted the dependency to 3.0.0-1.17 and it works again. Could it be related to the introduction of the DynamicKafkaSource in 3.1.0? * Factory class for the DynamicKafkaSource components. <a * href=https://cwiki.apache.org/confluence/x/CBn1D>FLIP-246: DynamicKafkaSource</a> * * <p>This source's key difference from {@link KafkaSource} is that it enables users to read * dynamically, which does not require job restart, from streams (topics that belong to one or more * clusters). If using {@link KafkaSource}, users need to restart the job by deleting the job and * reconfiguring the topics and clusters. We are still using KafkaSource in our pipelines. Best, Dominik Bünzli Data, Analytics & AI Engineer III From: Schwalbe Matthias <matthias.schwa...@viseca.ch> Date: Tuesday, 3 September 2024 at 09:59 To: Bünzli Dominik, INI-DNA-INF <dominik.buen...@swisscom.com>, user@flink.apache.org <user@flink.apache.org> Subject: RE: Kafka connector exception restarting Flink 1.19 pipeline Be aware: This is an external email. Hi Dominik, No clue why this happens, but it looks like that when restarting from the savepoint it uses the flink-connector-kafka version from your docker image (3.0.x ?) instead of the newer one you configured. How did you integrate the newer version? Thias From: dominik.buen...@swisscom.com <dominik.buen...@swisscom.com> Sent: Monday, September 2, 2024 1:35 PM To: user@flink.apache.org Subject: [External] Kafka connector exception restarting Flink 1.19 pipeline ⚠EXTERNAL MESSAGE – CAUTION: Think Before You Click ⚠ Dear Flink community We recently migrated our pipelines from Flink 1.17 to 1.19.0 (and subsequently to 1.19.1). We are sourcing events from Kafka and write enriched events back to Kafka. I’m currently using the flink-connector-kafka (3.2.0-1.19). When initially deploying (via k8s operator), the pipeline starts as expected and is generating output events that are written to Kafka. As soon as I deploy a new pipeline and perform a restart via savepoint sometime afterwards, the following error is thrown. Has anyone seen this before? Any idea how to solve the issue? 11:20:41.934 [jobmanager-io-thread-1] INFO o.a.f.r.o.c.RecreateOnResetOperatorCoordinator - Resetting coordinator to checkpoint. 11:20:41.935 [Thread-15] INFO o.a.f.r.s.c.SourceCoordinator - Closing SourceCoordinator for source Source: Kafka Source XYZ. 11:20:41.936 [Thread-15] INFO o.a.f.r.s.c.SourceCoordinator - Source coordinator for source Source: Kafka Source XYZ closed. 11:20:41.939 [jobmanager-io-thread-1] INFO o.a.f.r.s.c.SourceCoordinator - Restoring SplitEnumerator of source Source: Kafka Source XYZ from checkpoint. 11:20:41.942 [jobmanager-io-thread-1] ERROR o.a.f.r.o.c.RecreateOnResetOperatorCoordinator - Failed to reset the coordinator to checkpoint and start. java.io.IOException: The bytes are serialized with version 2, while this deserializer only supports version up to 1 at org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumStateSerializer.deserialize(KafkaSourceEnumStateSerializer.java:83) at org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumStateSerializer.deserialize(KafkaSourceEnumStateSerializer.java:43) at org.apache.flink.runtime.source.coordinator.SourceCoordinator.deserializeCheckpoint(SourceCoordinator.java:556) at org.apache.flink.runtime.source.coordinator.SourceCoordinator.resetToCheckpoint(SourceCoordinator.java:451) Kind regards Dominik Bünzli Data, Analytics & AI Engineer III Diese Nachricht ist ausschliesslich für den Adressaten bestimmt und beinhaltet unter Umständen vertrauliche Mitteilungen. Da die Vertraulichkeit von e-Mail-Nachrichten nicht gewährleistet werden kann, übernehmen wir keine Haftung für die Gewährung der Vertraulichkeit und Unversehrtheit dieser Mitteilung. Bei irrtümlicher Zustellung bitten wir Sie um Benachrichtigung per e-Mail und um Löschung dieser Nachricht sowie eventueller Anhänge. Jegliche unberechtigte Verwendung oder Verbreitung dieser Informationen ist streng verboten. This message is intended only for the named recipient and may contain confidential or privileged information. As the confidentiality of email communication cannot be guaranteed, we do not accept any responsibility for the confidentiality and the intactness of this message. If you have received it in error, please advise the sender by return e-mail and delete this message and any attachments. Any unauthorised use or dissemination of this information is strictly prohibited.
