one issue we are facing when upgraded to flink kubernetes operator 1.14.0 is in case of error in job manager, pod is showing healthy in argo cc @Gyula Fóra

Hi @Martijn Visser, Thanks for the hard work on flink-connector-kafka. I just wanted to know if there is ny tentative release date for its new version as some issues for dynamic kafka source has already been fixed in the main brain and also currently there is no compatible version of kafka connector for flink 2.2

Hello. Quesiton - is it possible to stagger tumbling window fire across keys. I have ~64M keyed windows on a 60-min tumbling window. All fire at once when watermark advances. I want to try to spread this out. To be clear — I need aligned window boundaries (same hour for all keys). I just want to stagger when the results are emitted, not the window boundaries themselves. Wrote a custom trigger that delays firing with a random processing-time timer:

public class JitteredEventTimeTrigger extends Trigger<Object, TimeWindow> {
      private final long maxJitterMillis;

      @Override
      public TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) {
          ctx.registerEventTimeTimer(window.maxTimestamp());
          return TriggerResult.CONTINUE;
      }

      @Override
      public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) {
          long jitter = ThreadLocalRandom.current().nextLong(maxJitterMillis);
          ctx.registerProcessingTimeTimer(ctx.getCurrentProcessingTime() + jitter);
          return TriggerResult.CONTINUE;
      }

      @Override
      public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) {
          return TriggerResult.FIRE;
      }

      @Override
      public void clear(TimeWindow window, TriggerContext ctx) {
          ctx.deleteEventTimeTimer(window.maxTimestamp());
      }
  }

Not seeing any output. I think WindowOperator https://github.com/apache/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java#L485-L487 might be cleaning up window state at maxTimestamp + allowedLateness(0) in the same onEventTime call - before my processing-time timer gets a chance to fire. Can anyone confirm? Flink 2.2 running on AWS Managed Flink

Hello, I have a question about checkpointing. If I want to disable checkpointing in flink-sql-client, how can I achieve this? I’ve tried several approaches, but after a few attempts it feels like checkpointing cannot be fully disabled at all. Flink version 2.2.1

Flink SQL> SET 'execution.checkpointing.interval' = 0;
[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.sql.parser.impl.ParseException: Encountered "0" at line 1, column 42.
Was expecting one of:
    <BINARY_STRING_LITERAL> ...
    <QUOTED_STRING> ...
    <PREFIXED_STRING_LITERAL> ...
    <UNICODE_STRING_LITERAL> ...
    <C_STYLE_ESCAPED_STRING_LITERAL> ...
    <BIG_QUERY_DOUBLE_QUOTED_STRING> ...
    <BIG_QUERY_QUOTED_STRING> ...

Flink SQL> SET 'execution.checkpointing.interval' = 'false';
[ERROR] Could not execute SQL statement. Reason:
java.lang.NumberFormatException: text does not start with a number, and is not a valid ISO-8601 duration format: false

Flink SQL> SET 'execution.checkpointing.interval' = '0s';
[ERROR] Could not execute SQL statement. Reason:
java.lang.IllegalArgumentException: Checkpoint interval must be larger than or equal to 10 ms

Flink SQL> SET 'execution.checkpointing.interval' = '0 s';
[ERROR] Could not execute SQL statement. Reason:
java.lang.IllegalArgumentException: Checkpoint interval must be larger than or equal to 10 ms

Flink SQL> SET 'execution.checkpointing.interval' = NONE;
[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.sql.parser.impl.ParseException: Encountered "NONE" at line 1, column 42.
Was expecting one of:
    <BINARY_STRING_LITERAL> ...
    <QUOTED_STRING> ...
    <PREFIXED_STRING_LITERAL> ...
    <UNICODE_STRING_LITERAL> ...
    <C_STYLE_ESCAPED_STRING_LITERAL> ...
    <BIG_QUERY_DOUBLE_QUOTED_STRING> ...
    <BIG_QUERY_QUOTED_STRING> ...

Flink SQL> SET 'execution.checkpointing.interval' = NULL;
[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.sql.parser.impl.ParseException: Encountered "NULL" at line 1, column 42.
Was expecting one of:
    <BINARY_STRING_LITERAL> ...
    <QUOTED_STRING> ...
    <PREFIXED_STRING_LITERAL> ...
    <UNICODE_STRING_LITERAL> ...
    <C_STYLE_ESCAPED_STRING_LITERAL> ...
    <BIG_QUERY_DOUBLE_QUOTED_STRING> ...
    <BIG_QUERY_QUOTED_STRING> ...

Hey folks - hoping someone can shed light on a weird MSK lag behavior I'm seeing. Setup: Managed Flink on AWS, KafkaSource, 256 partitions, parallelism 16, checkpoints every 1 min (~2-3s duration, no failures, ~700MB state). The problem: Flink's records_lag_max is basically flat at 200-1000 — app is caught up and happy. The only spikes are brief jumps to ~20K when windows fire, which is expected. But MSK's SumOffsetLag and MaxOffsetLag show a sawtooth pattern: lag grows linearly for ~50 minutes (up to ~13M sum / ~60K max), then drops abruptly to near zero. Repeats endlessly. This makes no sense to me. If offsets are committed on checkpoint (every minute), why would broker-side lag accumulate for 50 minutes before resetting? The 50-min cycle doesn't correspond to anything in my app config. Has anyone seen this? Is there something about how the Flink Kafka connector commits offsets that could explain this? Or is MSK's lag metric calculation just weird/delayed? Any pointers appreciated 🙏

Hello, I noticed there was an intention to release a new version of the Flink connector for Pulsar (v4.2.0): https://lists.apache.org/thread/bwd91nbcyflbhvhrm33qg7tjksl6b03w However, no binding votes were cast to approve or reject the request — only non-binding votes were submitted. What would be the next steps here? Thanks!

Hi team I am working on a problem where we need to dynamically decrease/increase the sources that a flink application is listening to based on different modes. Currently doing this leads to an error like

Cannot map checkpoint/savepoint state for operator f17fc775f6e82bc8d571beb78ac6c693 to the new program, because the operator is not available in the new program. If you want to allow to skip this, you can set the --allowNonRestoredState option on the CLI.

I want to avoid the usage of

allowNonRestoredState

flag since it could hide real issues What is the way to achieve starting from snapshots safely after a change in operator chain/sources?

Hello, I need to consume data from a avro-confluent topic created by KSQL using KSQL's atomic key format. The key is a primitive "int" no record hence I can't deserialize it with flink using the avro-confluent key format. Does anyone of you ever encountered this problem and how did you solve it? Thanks a lot for your help!

Hey all 👋 — sorry for the vague question and the cross-post since I also posted this to the mailing list. I wanted to bring up the topic here to see if anyone in this channel may have some insights to share. I’m running Flink 2.2 .1 K8s Operator 1.15 with the ForSt async state backend, checkpoints in S3, working state around 6TB. With

upgradeMode

last-state

and using the adaptive scheduler we observe: • Downscale rescales (e.g. parallelism 24 -> 12) take roughly 2 hours from spec change to RUNNING (no events are processed in this time • Upscale rescales in the same flow are much faster, taking around 12 minutes The relevant ForSt / checkpointing settings are: •

use-ingest-db-restore-mode: true

•

incremental-restore-async-compact-after-rescale: true

•

state-recovery.claim-mode: CLAIM

•

NATIVE

savepoint format, incremental + compressed checkpoints •

pipeline.max-parallelism: 120

• ForSt cache

size-based-limit

350GB on a 500Gi gp3 volume per TM. • I’m running one slot per TM and have tried parallelisms between 12 and 24 My questions are: • Is ~2h for a downscale at this state size in line with what others see, or does it suggest a misconfiguration? Most of the time appears to be in restore. • Is the asymmetry where upscales are much faster also expected? • Given the asymmetry, are there settings or patterns that make autoscaling viable at ~1TB state beyond adjusting

job.autoscaler.scale-down.max-factor

and

scale-down.interval

? Or is the practical recommendation at this scale to keep parallelism static and rescale manually during low-traffic windows? Thank you!

Hello people, Sorry in advance if this is not the right place for asking this and please let me know if i should move it but, Is anyone aware of when is planed for the Flink-operator to get Java 21 support? or if it its gonna even happen? 😅

Hey y'all, I'm trying to work build / run tests on Flink locally and I'm having issues, I think with my company's security software, because the

pekko

tests try to connect to my machine's local-but-external ip (ie

10.0.0.whatever

.) Is there a property or something else I can just

-D

in order to force pekko to use localhost instead of whatever discovery mechanism it's doing?

Hello everyone! I am currently migrating a Flink application (Java) from EMR to AWS EKS. We currently save checkpoints using RocksDB in HDFS. I am wondering if somebody has a recommendation or could point me to some doc in which I could see options and best practices to setup checkpoint store in EKS or K8S in general? We previously used S3 for checkpoint but the context I get is that there was some throttling and the team decided to move to RocksDB (not much info in that front unfortunately). Any help is appreciated! Thanks in advance!

Hello team, we are running into a correctness bug caused by flink k8s operator + autoscaling. Specifically, when kubernetes.operator.job.drain-on-savepoint-deletion is set to true, the operator uses a draining savepoint to stop the pipeline, then resumes from it with modified parallelism overrides. I think this is wrong behavior. Per Flink document, "If you want to resume the job at a later point in time, then do not drain the pipeline because it could lead to incorrect results when the job is resumed". I just want to confirm this is indeed unintended behavior. I can contribute a PR.

Hi team, anyone got this error when switching to native s3 fs with forst state backend?

AsyncException{org.forstdb.RocksDBException: Exception when Flush file, path: <s3://yotpo-cdp/flinkcdp/checkpoints/shared/op_AsyncKeyedProcessOperator_f199ad3ac3e86e5fc9372ebb239a06cb__33_64__attempt_0/db/260675.sst>}
	at org.apache.flink.runtime.asyncprocessing.operators.AbstractAsyncKeyOrderedStreamOperator.handleAsyncException(AbstractAsyncKeyOrderedStreamOperator.java:120)
	at org.apache.flink.core.asyncprocessing.AsyncFutureImpl.completeExceptionally(AsyncFutureImpl.java:553)
	at org.apache.flink.state.forst.ForStDBPutRequest.completeStateFutureExceptionally(ForStDBPutRequest.java:87)
	at org.apache.flink.state.forst.ForStWriteBatchOperation.lambda$process$0(ForStWriteBatchOperation.java:80)
	at java.base/java.util.concurrent.CompletableFuture$AsyncRun.run(Unknown Source)
	at org.apache.flink.util.concurrent.DirectExecutorService.execute(DirectExecutorService.java:247)
	at java.base/java.util.concurrent.CompletableFuture.asyncRunStage(Unknown Source)
	at java.base/java.util.concurrent.CompletableFuture.runAsync(Unknown Source)
	at org.apache.flink.state.forst.ForStWriteBatchOperation.process(ForStWriteBatchOperation.java:65)
	at org.apache.flink.state.forst.ForStStateExecutor.lambda$executeBatchRequests$2(ForStStateExecutor.java:201)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
	at org.apache.flink.state.forst.ForStStateExecutor$ForStExecutorThreadFactory.lambda$newThread$0(ForStStateExecutor.java:338)
	at java.base/java.lang.Thread.run(Unknown Source)
Caused by: org.forstdb.RocksDBException: Exception when Flush file, path: <s3://yotpo-cdp/flinkcdp/checkpoints/shared/op_AsyncKeyedProcessOperator_f199ad3ac3e86e5fc9372ebb239a06cb__33_64__attempt_0/db/260675.sst>
	at org.forstdb.RocksDB.write0(Native Method)
	at org.forstdb.RocksDB.write(RocksDB.java:1843)
	at org.apache.flink.state.forst.ForStDBWriteBatchWrapper.flush(ForStDBWriteBatchWrapper.java:136)
	at org.apache.flink.state.forst.ForStDBWriteBatchWrapper.flushIfNeeded(ForStDBWriteBatchWrapper.java:155)
	at org.apache.flink.state.forst.ForStDBWriteBatchWrapper.put(ForStDBWriteBatchWrapper.java:117)
	at org.apache.flink.state.forst.ForStDBPutRequest.process(ForStDBPutRequest.java:68)
	at org.apache.flink.state.forst.ForStWriteBatchOperation.lambda$process$0(ForStWriteBatchOperation.java:70)
	... 10 more
	Suppressed: org.forstdb.RocksDBException: Exception when Flush file, path: <s3://yotpo-cdp/flinkcdp/checkpoints/shared/op_AsyncKeyedProcessOperator_f199ad3ac3e86e5fc9372ebb239a06cb__33_64__attempt_0/db/260675.sst>
		at org.forstdb.RocksDB.write0(Native Method)
		at org.forstdb.RocksDB.write(RocksDB.java:1843)
		at org.apache.flink.state.forst.ForStDBWriteBatchWrapper.flush(ForStDBWriteBatchWrapper.java:136)
		at org.apache.flink.state.forst.ForStDBWriteBatchWrapper.close(ForStDBWriteBatchWrapper.java:144)
		at org.apache.flink.state.forst.ForStWriteBatchOperation.lambda$process$0(ForStWriteBatchOperation.java:67)
		... 10 more

Hello , is the flink kubernetes operator 1.15.0 compatible with flink 1.17/1.18? https://flink.apache.org/2026/05/26/apache-flink-kubernetes-operator-1.15.0-release-announcement/#flink-22-compatibility The announcement does not mention them.

Hello everyone. Currently, I'm using Flink with the architecture Kafka -> Flink -> Starrocks Database. I'm having trouble using merge commits in Starrocks. I need the merge commit attribute to reduce the number of tablet versions in Starrocks, but when I use this attribute, I get missing records. For example, I push 500 records per second into Kafka for 10 seconds. This should result in 5000 records in Starrocks, but in reality, when I count the records, there aren't enough. Does anyone have any ideas? Sorry for my limited English.

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 !