[Feature Request] Flink 2.1.0 support - Getting "Unsupported version: 728" error

[rafael-gumiero]

Issue Type

• Bug Report
• Feature Request
• Question

Summary

Request for Flink 2.1.0 support and clarification on version compatibility

Description

We are experiencing state deserialization failures when attempting to use flink-connector-clickhouse-2.0.0:0.1.1 with Flink runtime 2.1.0.

According to the compatibility matrix, the connector currently supports up to Flink 2.0.x, but there's no mention of Flink 2.1.0 support.

Environment

Flink Configuration:

• Runtime Version: 2.1.0
• Connector Artifact: flink-connector-clickhouse-2.0.0:0.1.1:all
• Java Version: 21
• Deployment: Kubernetes with Flink Operator v2_1

Gradle Dependencies:

var flinkVersion = "2.1.0"
flinkShadowJar("com.clickhouse.flink:flink-connector-clickhouse-2.0.0:0.1.1:all")

Infrastructure:

• Kubernetes with Flink Operator
• RocksDB State Backend (30Gi ephemeral volumes)
• S3 checkpoint storage
• Checkpoint interval: 60s
• Parallelism: 10 (autoscaler up to 120)

Error Details

When the job attempts to restore from a checkpoint, it fails with:

org.apache.flink.util.FlinkRuntimeException: Failed to deserialize value
    at org.apache.flink.streaming.api.operators.util.SimpleVersionedListState$DeserializingIterator.next(SimpleVersionedListState.java:140)
    ...
Caused by: java.io.IOException: Unsupported version: 728
    at org.apache.flink.connector.clickhouse.sink.ClickHouseAsyncSinkSerializer.deserializeRequestFromStream(ClickHouseAsyncSinkSerializer.java:41)
    at org.apache.flink.connector.base.sink.writer.AsyncSinkWriterStateSerializer.deserialize(AsyncSinkWriterStateSerializer.java:81)

Root Cause Analysis

The error appears to be a serialization format incompatibility between Flink 2.0 and 2.1. The connector was built for Flink 2.0.x and doesn't support the state serialization changes in Flink 2.1.

Questions

1. Is Flink 2.1.0 support planned?

   • If yes, what is the expected timeline/milestone?
   • Should we expect it in the next minor release (e.g., 0.1.2)?

2. Is there a snapshot/beta version available that we could test with Flink 2.1?

Use Case Context

We're building a high-throughput streaming pipeline:

• Volume: ~700TB/month of log data
• Source: 70+ Kafka topics from multiple microservice clusters
• Transformation: JSON logs → OpenTelemetry format
• Sink: ClickHouse for real-time analytics
• Requirements: High availability, exactly-once semantics (if possible)

Request

Could you please provide:

• ✅ Confirmation of current Flink 2.1 support status
• ✅ Roadmap/timeline for Flink 2.1 support
• ✅ Recommended stable Flink version for production
• ✅ Any beta/snapshot versions available for testing
• ✅ Best practices for our high-volume use case

Additional Context

• This is a new project (no production checkpoints to preserve)
• We can test beta/snapshot versions in our QA environment
• We're willing to contribute feedback/testing for Flink 2.1 support

Related

• Flink 2.1.0 Release: https://flink.apache.org/news/2024/11/18/apache-flink-2.1.0-release-announcement/
• Our connector dependency: https://github.com/ClickHouse/flink-connector-clickhouse/blob/main/README.md#supported-flink-versions

---

Thank you for this excellent connector! Looking forward to your guidance. 🙏

[mzitnik]

Hi @rafael-gumiero
Is this happening on the first snapshot restore (or only if you scale and then do the restore)
Is it possible, for testing purposes, to disable checkpointing and see the outcome?
We have a test for 2.1.0, and the supported matrix is incorrect. I will update the README

[antomys]

Hello! flink-1.20.3. Using clickhouse 1.17 connection. Facing simillar issue:

org.apache.flink.util.FlinkRuntimeException: Failed to deserialize value
    at org.apache.flink.streaming.api.operators.util.SimpleVersionedListState$DeserializingIterator.next(SimpleVersionedListState.java:140)
    at java.base/java.util.Iterator.forEachRemaining(Unknown Source)
    at org.apache.flink.util.CollectionUtil.iterableToList(CollectionUtil.java:119)
    at org.apache.flink.streaming.runtime.operators.sink.StatefulSinkWriterStateHandler.createWriter(StatefulSinkWriterStateHandler.java:103)
    at org.apache.flink.streaming.runtime.operators.sink.SinkWriterOperator.initializeState(SinkWriterOperator.java:153)
    at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.initializeOperatorState(StreamOperatorStateHandler.java:147)
    at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:294)
    at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.initializeStateAndOpenOperators(RegularOperatorChain.java:106)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreStateAndGates(StreamTask.java:858)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$restoreInternal$5(StreamTask.java:812)
    at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreInternal(StreamTask.java:812)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:771)
    at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:970)
    at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:939)
    at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:763)
    at org.apache.flink.runtime.taskmanager.Task.run(Task.java:575)
    at java.base/java.lang.Thread.run(Unknown Source)
Caused by: java.io.IOException: Unsupported version: 1286
    at org.apache.flink.connector.clickhouse.sink.ClickHouseAsyncSinkSerializer.deserializeRequestFromStream(ClickHouseAsyncSinkSerializer.java:41)

Running in standalone mode in k8s. During leader election and job restarting, we are facing this exception, which makes job to fail