Hi, I'm new to Apache Flink and could use some help. I have a question: can a single TaskManager consume data from multiple Kafka topics? If so, when using RocksDB as the state backend, does it maintain state separately for each topic, or is the state shared across all topics processed by that TaskManager? Thanks in advance!

Hi, trying to understand state migration and

compatibleAfterMigration

. Initially my understanding was that the old serializer is used to read the stored bytes, the instance object is then written with the new serializer and these new bytes are read by the new serializer, in short I would have assumed that it ensured that the instances received in the operators are always constructed using the new serializer. But I'm hitting a bug where I suspect that the instance read by the old serializer is simply passed to the operator without any migration. (Note this is the operator state of the Async operator). Could someone clarify my understanding about state migration with

compatibleAfterMigration

? Thanks! 🙏

Hi everyone, I am using Apache Flink 1.20.4, and I have the following high-availability requirement: I want to configure my Flink cluster so that if a job is running on one node (for example, Node 1) and that node suddenly fails, another node in the cluster automatically takes over the job from the latest available state. The new node should have access to all the required checkpoint/savepoint data so that the job can resume processing without data loss or manual intervention. What is the recommended HA architecture and configuration to achieve this? Specifically: • Which HA service should I use? • Where should checkpoints/savepoints be stored? • Which state backend is recommended? I do not want to use Kubernetes or HDFS. I have seen recommendations to use GlusterFS as shared storage for checkpoints and savepoints. Is GlusterFS a recommended approach for this setup, or is there a better alternative? Any suggestions or best practices would be appreciated.

Hi everyone, We are using Managed Flink in AWS with the flink version 1.18.1. We want to upgrade our flink jobs to 2.x for java 17 support and more libraries etc, however our flink job has state that is serialized using the kryo serializer (since we are using object types that are not considered POJOs). The kryo serializer state is backward incompatible when upgrading from 1.18.x to 2.x since the kryo serializer version is different and they uses different binary compatibilities. (src: docs.aws.amazon.com/managed-flink/latest/java/state-compatibility.html#…) Is there any way to restore the state and still upgrade to 2.x, or the only way is to skip state restoration and rely on the source to build up the state? Or is there any way to change the serializer used in flink 1.18x before upgrading? Any inputs would be helpful Thank you in advance! P.S We cannot do state manipulation using the state processor api since the snapshots etc are managed by AWS and the absolute paths cannot be accessed

Hi everyone, I am using Apache Flink 1.20.4, and I have the following high-availability requirement: I want to configure my Flink cluster so that if a job is running on one node (for example, Node 1) and that node suddenly fails, another node in the cluster automatically takes over the job from the latest available state. The new node should have access to all the required checkpoint/savepoint data so that the job can resume processing without data loss or manual intervention. What is the recommended HA architecture and configuration to achieve this? Specifically: • Which HA service should I use? • Where should checkpoints/savepoints be stored? • Which state backend is recommended? I do not want to use Kubernetes or HDFS. I have seen recommendations to use GlusterFS as shared storage for checkpoints and savepoints. Is GlusterFS a recommended approach for this setup, or is there a better alternative? Any suggestions or best practices would be appreciated.

Hi team, are there any plans to upgrade Py4j to 0.10.9.9 for pyflink? nightlies.apache.org/flink/…/index.html#… Our team is running into some python dependency issues, as Py4j 0.10.9.7 was a relatively old version released in 2022.

Hey all 👋 — I sent this to the mailing list as well but I wanted to highlight the issue here as well since it could be a pretty big blocker 😅 : issues.apache.org/jira/browse/FLINK-37686 We ran into this last night and from what I can tell all impacted classes are unchanged across all Flink releases. I tried to collect my findings into the comment on this issue and I would be happy to help out with a fix here. Thanks!

Hi everyone, How do you handle schema evolution in Apache Flink? Are you using any specific tools, configurations, or best practices to manage schema changes effectively? Any suggestions or best practices would be appreciated.

Hi everyone, We're evaluating strategies for handling schema evolution in Flink SQL jobs. We tested both: • Running the old and new jobs in parallel, and • A savepoint-based approach using versioned tables/views. Both approaches worked but introduced challenges such as operator state mapping, handling state for newly added operators, multiple sink tables, Kafka retention limitations, deployment complexity, and deadlocks during migration. I'm interested in knowing how others handle schema evolution in production. What approach or tooling do you use to evolve Flink SQL jobs with minimal downtime while preserving state and ensuring data consistency? Any help would be highly appreciated😀

Hi #C03G7LJTS2G I have observed one issue/behaviour with Flink Iceberg streaming pipeline. I have checkpoint configured as 10 secs since the expected latency is 10 secs and I am able to get the records as expected but what I have seen is lot of Iceberg snapshots are also created for that table even if there is no data flowing in every 10 seconds. Checkpoints are working fine, no errors but I was expecting that if there is new data then only the snapshot should be written. So what am I missing here? Is this the expected behavior? Also, the other problem is that, I only inserted 3 records in my topic and the same can be seen in the target table but in the Flink UI I can see something as attached. Why am I seeing this?

Hi #C03G7LJTS2G, I am running a kafka to iceberg writer for events data and see a consistent consumer lag, but from querying kafka I can see that the data is actually there, so flink is actually delaying commiting offset to kafka. I can see a correlation between checkpoint interval and consumer lag. Is there a way to make it zero ? or is this expected behaviour ?

Hi all , question on unbounded aggregations in Flink: I need to compute all-time metrics (e.g. count distinct, leaderboards) rather than windowed ones. The straightforward approach is to keep everything in keyed state, but I'm worried about state growing unbounded pretty quickly. Is there an established pattern for this? One option I was considering was to use async I/O against our DB to check whether a key/combination already exists, and computing the distinct counts from there instead of holding it all in Flink state. (not ideal) Curious to hear if there's a better approach (approximate sketches, external store, something else). gratitude thank you

Hello all! Question about the Flink SQL Client: I've set up a fluss cluster where flink is needed. To make quick queries on the database, I've tried using the flink sql client. But queries only works when I use "LIMIT 10". If I don't, I get this error:

[ERROR] Could not execute SQL statement. Reason:
java.lang.ClassNotFoundException: org.apache.iceberg.data.IcebergGenericReader
        ... 15 more

And the full error in the jobmanager out file:

java.lang.RuntimeException: One or more fetchers have encountered exception
  at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager.checkErrors(SplitFetcherManager.java:333) ~[flink-connector-files-1.20.0.jar:1.20.0]
  at org.apache.flink.connector.base.source.reader.SourceReaderBase.getNextFetch(SourceReaderBase.java:228) ~[flink-connector-files-1.20.0.jar:1.20.0]
  at org.apache.flink.connector.base.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:190) ~[flink-connector-files-1.20.0.jar:1.20.0]
  at org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:422) ~[flink-dist-1.20.0.jar:1.20.0]
  at org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:68) ~[flink-dist-1.20.0.jar:1.20.0]
  at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65) ~[flink-dist-1.20.0.jar:1.20.0]
  at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:638) ~[flink-dist-1.20.0.jar:1.20.0]
  at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:231) ~[flink-dist-1.20.0.jar:1.20.0]
  at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:973) ~[flink-dist-1.20.0.jar:1.20.0]
  at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:917) ~[flink-dist-1.20.0.jar:1.20.0]
  at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:970) ~[flink-dist-1.20.0.jar:1.20.0]
  at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:949) ~[flink-dist-1.20.0.jar:1.20.0]
  at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:763) ~[flink-dist-1.20.0.jar:1.20.0]
  at org.apache.flink.runtime.taskmanager.Task.run(Task.java:575) ~[flink-dist-1.20.0.jar:1.20.0]
  at java.lang.Thread.run(Thread.java:1583) ~[?:?]
Caused by: java.lang.NoClassDefFoundError: org/apache/iceberg/data/IcebergGenericReader
  at org.apache.fluss.lake.iceberg.source.IcebergRecordReader.<init>(IcebergRecordReader.java:65) ~[fluss-lake-iceberg-0.9.1-incubating.jar:0.9.1-incubating]
  at org.apache.fluss.lake.iceberg.source.IcebergLakeSource.createRecordReader(IcebergLakeSource.java:100) ~[fluss-lake-iceberg-0.9.1-incubating.jar:0.9.1-incubating]
  at org.apache.fluss.flink.lake.reader.LakeSnapshotScanner.pollBatch(LakeSnapshotScanner.java:54) ~[fluss-flink-1.20-0.9.1-incubating.jar:0.9.1-incubating]
  at org.apache.fluss.flink.source.reader.BoundedSplitReader.pollBatch(BoundedSplitReader.java:128) ~[fluss-flink-1.20-0.9.1-incubating.jar:0.9.1-incubating]
  at org.apache.fluss.flink.source.reader.BoundedSplitReader.poll(BoundedSplitReader.java:121) ~[fluss-flink-1.20-0.9.1-incubating.jar:0.9.1-incubating]
  at org.apache.fluss.flink.source.reader.BoundedSplitReader.readBatch(BoundedSplitReader.java:76) ~[fluss-flink-1.20-0.9.1-incubating.jar:0.9.1-incubating]
  at org.apache.fluss.flink.source.reader.FlinkSourceSplitReader.fetch(FlinkSourceSplitReader.java:154) ~[fluss-flink-1.20-0.9.1-incubating.jar:0.9.1-incubating]
  at org.apache.flink.connector.base.source.reader.fetcher.FetchTask.run(FetchTask.java:58) ~[flink-connector-files-1.20.0.jar:1.20.0]
  at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:165) ~[flink-connector-files-1.20.0.jar:1.20.0]
  at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.run(SplitFetcher.java:117) ~[flink-connector-files-1.20.0.jar:1.20.0]
  at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:572) ~[?:?]
  at java.util.concurrent.FutureTask.run(FutureTask.java:317) ~[?:?]
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144) ~[?:?]
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642) ~[?:?]
  ... 1 more
Caused by: java.lang.ClassNotFoundException: org.apache.iceberg.data.IcebergGenericReader
  at jdk.internal.loader.BuiltinClassLoader.loadClass(BuiltinClassLoader.java:641) ~[?:?]
  at jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(ClassLoaders.java:188) ~[?:?]
  at java.lang.ClassLoader.loadClass(ClassLoader.java:526) ~[?:?]
  at org.apache.fluss.lake.iceberg.source.IcebergRecordReader.<init>(IcebergRecordReader.java:65) ~[fluss-lake-iceberg-0.9.1-incubating.jar:0.9.1-incubating]
  at org.apache.fluss.lake.iceberg.source.IcebergLakeSource.createRecordReader(IcebergLakeSource.java:100) ~[fluss-lake-iceberg-0.9.1-incubating.jar:0.9.1-incubating]
  at org.apache.fluss.flink.lake.reader.LakeSnapshotScanner.pollBatch(LakeSnapshotScanner.java:54) ~[fluss-flink-1.20-0.9.1-incubating.jar:0.9.1-incubating]
  at org.apache.fluss.flink.source.reader.BoundedSplitReader.pollBatch(BoundedSplitReader.java:128) ~[fluss-flink-1.20-0.9.1-incubating.jar:0.9.1-incubating]
  at org.apache.fluss.flink.source.reader.BoundedSplitReader.poll(BoundedSplitReader.java:121) ~[fluss-flink-1.20-0.9.1-incubating.jar:0.9.1-incubating]
  at org.apache.fluss.flink.source.reader.BoundedSplitReader.readBatch(BoundedSplitReader.java:76) ~[fluss-flink-1.20-0.9.1-incubating.jar:0.9.1-incubating]
  at org.apache.fluss.flink.source.reader.FlinkSourceSplitReader.fetch(FlinkSourceSplitReader.java:154) ~[fluss-flink-1.20-0.9.1-incubating.jar:0.9.1-incubating]
  at org.apache.flink.connector.base.source.reader.fetcher.FetchTask.run(FetchTask.java:58) ~[flink-connector-files-1.20.0.jar:1.20.0]
  at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:165) ~[flink-connector-files-1.20.0.jar:1.20.0]
  at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.run(SplitFetcher.java:117) ~[flink-connector-files-1.20.0.jar:1.20.0]
  at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:572) ~[?:?]
  at java.util.concurrent.FutureTask.run(FutureTask.java:317) ~[?:?]
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144) ~[?:?]
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642) ~[?:?]
  ... 1 more

Any ideas on what the problem could be? I'm happy to provide any other information if necessary.

Hello everyone. I am currently working on Flink 2.2 with Kubernetes. I was wondering if there was a plan to upgrade hadoop version to 3.4.1 and more recent to use azure workload identity ? Thank you in advance !

Hello everyone, Following our recent upgrade from Flink 1.19 to 2.2, we have noticed a memory leak that seems to originate from FLIP-513 (Split-level Watermark Metrics). When consuming from Paimon/S3 (which frequently creates new splits, ~9,000 splits/hour in our case), Flink instantiates metric groups per split (

SourceSplitMetricGroup

), but these metric groups are never released after the split finishes. Over a few days, this fills up the heap, triggers heavy GC saturation, causes source starvation, and increases consumer lag until we restart the job. Our heap dump shows that 99.9% of the leaked metric groups contain the

'split'

key. Is there already a JIRA issue or a PR tracking this lifecycle cleanup bug for FLIP-513? If not, we can create a JIRA issue with our heap dump findings. Also, is there a recommended configuration workaround to completely disable/exclude these split-level watermark metrics in Flink 2.2 to prevent registration in the meantime? Thanks!

Hi everyone, cross-posting my email here. We're running a Flink 1.20.5 job with a Kafka source (KafkaSource / flink-connector-base) that might hit FLINK-37663, a lost-wakeup in

FutureCompletingBlockingQueue.waitOnPut()

that permanently stalls a fetcher thread under sustained backpressure after a mass split-assignment burst. The ticket shows this is resolved upstream, with fix versions including 1.20.6, 2.0.3, 2.1.4, 2.2.2, 2.3.0, and 2.4.0. Only 2.3.0 has actually shipped so far. A couple of questions for the community: 1. Is there a workaround for this that doesn't require a full Flink version bump? 2. Is there a rough timeline for a 1.20.6 patch release, or is upgrading to 2.3.0 the recommended path forward? Thanks in advance.

Hello all! I have a PR in to fix a bug with the python handling of

JdbcSink

from the JDBC Connector. Can someone take a look at it for me? Happy to make any changes to it too. github.com/apache/flink/pull/27528

Hey all, I’m running into an issue. We’re using

flink-cdc-pipeline-connector-iceberg (3.6.0-2.2)

to stream MySQL → Iceberg on Glue. When the partition key column is updated, we’re getting duplicate rows instead of the row being moved to the new partition. Repro:

CREATE TABLE t (
  id BIGINT,
  val VARCHAR,
  region VARCHAR
) PARTITIONED BY identity(region);

INSERT id=1, region=north;

UPDATE id=1 SET region=east;

After the update,

id=1

appears twice: one row with

region=north

(old/stale) and one with

region=east

(new). From what I understand, this happens because the equality delete only uses the primary key and doesn’t include the partition column, so the delete doesn’t remove the row from the old partition. Has anyone solved this for partitioned tables? Is scheduled compaction the only practical workaround, or is there a write-side fix? Also, does

flink-cdc-pipeline-connector-iceberg

expose any configuration to set the equality fields? Any pointers would be appreciated.

Looking at PubSub as a source for a Flink 2.x job, and the connector story looks stalled. What should we expect for the future of this connector? Is a Flink 2.x port on anyone's roadmap, and is there any sense of when it might land?

Hey team 👋 Is it possible to release the latest flink-kubernetes-operator 1.16.0 and release-1.16.1-rc1 tag to docker hub ? 🙏 We've found some CVEs reported in the 1.15.0 version and would like to bump the image asap 🙂 The rc1 version also fixes the application ID , would be great to also test the fix 🙂 cc @Marton Balassi 🙏

Hi Team, Flink supports mTLS today - nightlies.apache.org/flink/…/security-ssl - has anyone tried using Istio/Envoy for handling internal connectivity certificates as well?

Hey Team 👋 Looking for some guidance from the flink community on managing flink deployments on Kubernetes. We recently migrated our flink workloads from a Hadoop VM based cluster to a Kubernetes Operator-managed deployment. We are currently observing high memory consumption in the Flink containers. Our containers also run additional observability agents, which require their own memory overhead. The challenge is that when we configure the total process memory for the Jobmanager and Taskmanager, the resulting container sizing doesn’t leave enough headroom for these additional agents. Has anyone faced a similar situation or have recommendations on how to configure flink memory/process memory. Any guidance or best practices would be greatly appreciated. Thanks!

Hi, has anyone tried using the Iceberg dynamic sink with a VARIANT column? We’re using Flink 2.2 and haven’t been able to get it working. Has anyone run into this or found a workaround?

Can anyone please help - https://github.com/apache/iceberg/pull/17631 Any tentative timeline on when this PR can be merged ?

Hi Team , Is there a way can define the container Memory 4GB and Total Process Memory to 3GB . And if we increase the container memory 4 --> 6 but still my Total process Memory should remain to 3GB

Hi team, We’re currently exploring Flink CDC for ingesting data into Iceberg, and one of the areas we’re investigating is compaction. The compaction approach we currently use for Spark-based ingestion doesn’t seem to work efficiently here, mainly because of frequent commits and continuous ingestion, which result in a large number of small files. I wanted to understand what the standard approach is for handling compaction in this scenario. A couple of approaches we’re considering are: • Delayed compaction: Leave recent partitions untouched and compact only partitions that are at least X days old, as they’re less likely to receive new data. • Branch-based approach: Ingest data into one Iceberg branch and perform compaction and reads on another branch. Has anyone implemented either of these approaches, or found a more effective way to handle compaction with continuous Flink CDC ingestion? Please share your experiences or suggestions. Thanks!

Hi everyone, Has anyone benchmarked memory usage for a MySQL → Flink CDC → Iceberg pipeline at scale? I’m trying to establish a baseline for estimating Flink TaskManager memory requirements per table, considering factors such as: • Number of columns and their data types • Peak CDC throughput (records/sec) I’m curious how others in production or large-scale deployments approach this. Do you use any rule of thumb or estimation model based on these metrics, or is load testing the only reliable way to determine memory requirements? If anyone has done similar benchmarking or has shared benchmark results, it would be great to hear about your approach and findings. Thanks!

Hi , if I have multiple jobs in operator mode there all and end up with the same context URL, so how can I configure context path for different different JOBS? So is there any reverse proxy available? We are using flink operator in openshift cluster It is community driven operator

Hi everyone, I’m migrating a Flink elastic search storage job from the ES7 connector to the ES8 connector and need guidance on failure handling. Setup: • Flink 2.2.1, Java 21. • Existing:

flink-connector-elasticsearch7:4.0.0-2.0

, client 7.10.2. • New:

flink-connector-elasticsearch8:4.0.0-2.0

, overriding the Java client to 8.19.22.

.setBulkResponseInspectorFactory(new ElasticsearchFailureHandler(host))
.setBulkFlushBackoffStrategy(FlushBackoffType.EXPONENTIAL, maxRetries, delay)

Our custom inspector classified failed bulk items and decided whether to ignore selected permanent document errors, perform restricted recovery, or fail the task. With

Elasticsearch8AsyncSinkBuilder

, I cannot find equivalent failure-handler or bounded-retry options. The released writer appears to requeue all failed bulk items without distinguishing permanent errors from transient ones or limiting attempts. Permanent failures could therefore retry indefinitely; task restarts can also replay the same problematic records. Is there a supported replacement/extension point for error classification, bounded retries/backoff, and durable DLQ handling, without maintaining a custom sink/writer? Was this intentionally changed with the async-sink architecture, or is it a feature-parity gap? I found FLINK-35546 and PR #107. Are these the right items to follow? We want to quarantine individual bad records or run time exceptions from elastic search such as index closed , shrink exceptions safely.

Hi all 👋 Hitting a startup failure with Flink CDC 3.7.0 + StarRocks sink on Flink 2.2 — wondering if anyone else has seen it or if it's a known issue. Setup • Flink 2.2.1 standalone session cluster, Java 17 • Flink CDC 3.7.0-2.2 (bin tarball + pipeline connectors from Maven Central) • StarRocks 4.1 sink, Postgres and MySQL sources • Same pipelines/cluster work fine on 3.6.0-2.2 Problem Any pipeline using the StarRocks sink fails as soon as the sink writer is created, even on a brand-new job (no savepoint): java.lang.NoSuchMethodError: 'org.apache.flink.api.connector.sink2.SinkWriter org.apache.flink.api.connector.sink2.StatefulSink.createWriter(org.apache.flink.api.connector.sink2.Sink$InitContext)' at org.apache.flink.api.connector.sink2.StatefulSink.createWriter(StatefulSink.java:41) at org.apache.flink.streaming.runtime.operators.sink.StatefulSinkWriterStateHandler.createWriter(StatefulSinkWriterStateHandler.java:122) at org.apache.flink.streaming.runtime.operators.sink.SinkWriterOperator.initializeState(SinkWriterOperator.java:171) at org.apache.flink.cdc.runtime.operators.sink.DataSinkWriterOperator.initializeState(DataSinkWriterOperator.java:133) What I noticed flink-cdc-dist-3.7.0-2.2.jar bundles a new sink2 compat class

StatefulSinkWriterAdapter

that 3.6.0-2.2 didn't have, so it looks like the updated compatibility shim doesn't line up with the StarRocks sink on Flink 2.2. Why it matters We need 3.7.0 for FLINK-38742 (Postgres TIMESTAMPTZ → TIMESTAMP_LTZ); on 3.6.0 the StarRocks sink rejects TIMESTAMP_TZ columns. Questions: 1. Is this a known issue / is a 3.7.1 fix planned? 2. Any workaround besides backporting #4181 onto 3.6?