Hello 👋🏻 I recently got community review and approval for https://github.com/apache/flink/pull/27598 What are the next steps? I suppose someone with write permissions need to look at it. But should I communicate it further?

Hi all, https://github.com/apache/flink/pull/27784 All checks have passed successfully. I would appreciate it if someone could take a look and provide feedback or approval. Thank you!

@Shaun Wanford has left the channel

Hi there. Recently we were faced by two issues we identified in the HybridSource (FLIP-150), they are: 1. We found there is a possibility of lost splits if there is a snapshot and crash during source switch. a. At this part, when snapshot state, if currentReader is null, it will return empty list directly. b. So when there is a recovery, where the job recovers from a checkpoint (then the splits from checkpoint was put to restoredSplits field) c. Then there is a savepoint/checkpoint happening, however, the currentReader may still be null because the reader just requested a new enumerator. d. So the empty list is snapshot as state, but then the job crashes again immediately. e. Upon the next recovery, the job no longer see the last restored splits, they were cleared due to last checkpoint. f. (although this seems to be a very rare situation but we seems indeed has encountered once). 2. The second is the same as FLINK-30514, the

start()

method was called before

restoredSplits

was set to the underlying readers. a. This is critical to us because we found this behavior is in-compatible with the Iceberg source reader because it request split in the

start

method b. In the iceberg source, on start it checks if there is recovered splits, but because of the order of calling

start

in the HybridSource, it never finds the recovered splits. c. this will become a issue because the reader will double request splits on recovery, this messed up watermark if we enabled watermark for the iceberg reader (multiple splits was assigned to one reader at the sametime, the second will block watermarks from emitting).

how do I fix this issue?

org.rocksdb.RocksDBException: While open a file for appending: /opt/flink/log/tmp_tm_flink-native-concurrent-counter-fingerprint-taskmanager-2-176_tmp_job_fdfa20c065c3443a0f05d56a45c312c3_op_KeyedProcessOperator_58a461be943b6304e654ee058d1ec08e__7_10__uuid_2eebc0b8-34d8-4bb8-9651-ba79bc3ef8bf_bfeb6a0e-20fd-4e91-835f-7c6616081783_LOG: File name too long
	at org.rocksdb.RocksDB.open(Native Method)
	at org.rocksdb.RocksDB.open(RocksDB.java:315)
	at org.apache.flink.state.rocksdb.RocksDBOperationUtils.openDB(RocksDBOperationUtils.java:89)
	... 28 more

I don't seem to have control over the file name and the operator name is auto generated (flink sql)

Has anyone been able to get Flink to work writing checkpoints to Azure Blob Store using a Workload Identity? I'm looking for a working example as I seem to keep getting this error:

Failed to initialize class org.apache.hadoop.fs.azurebfs.oauth2.WorkloadIdentityTokenProvider

even though the Workload Identity is correctly mapped into my pod.

Has anyone found an easy automatic workaround for https://issues.apache.org/jira/browse/FLINK-32212 ?

Message: java.lang.IllegalStateException: The library registration references a different set of library BLOBs than previous registrations for this job: ...

Hi all, I wanted to understand how PRs are currently prioritized and reviewed. I’ve noticed that several PRs remain open for a long time, often due to delayed or pending reviews. Is there a standard process in place for reviewing and merging PRs? Additionally, is there anything we can do as a community to help reduce the backlog and close open PRs more efficiently? I believe this would also help keep contributors motivated and engaged.

Hi guys I need your help here, I'm having trouble with state clean up when upgrading from Flink 1.16.3 to 1.20.3, using rocks dB with incremental false. In our code base we are managing our state by ouself, meaning clearing state when finishing some task, but in 1.20 3 the state keeps still without cleaning after task finishes I'm talking about 10 - 20 MB state (small)

It seems that there is an open bug on what I have described above, any idea how to solve it? https://issues.apache.org/jira/plugins/servlet/mobile#issue/FLINK-35853

Hi all! I have some questions about Flink enrichment patterns. We have real-time data that we need to join with 10+ tables, including historical data, but we don't have direct access to the SQL tables. A few questions: 1. What are the best practices for this kind of multi-table enrichment at scale? 2. Can we use our data warehouse as a source? We have CDC set up and are wondering if we can tap the JDBC connector for warehouses like BigQuery and Snowflake — has anyone tried this? 3. Are there other options we should consider? Any input appreciated!

Hi Flink team, I tried to send email to user-subscribe@flink.apache.org and subscribe, but I didn't get any confirmation email back. I checked spam folder as well, but nothing. Any ideas? Thanks! ☺️

Hi everyone, I’m trying to build a real-time data pipeline using Flink CDC. My goal is to capture CDC data from Oracle and load it into PostgreSQL. However, based on the documentation I’ve found so far, PostgreSQL seems to be supported only as a source, not as a sink. Is there any way to work around this or achieve the same result? Thanks a lot for your help!

Hi All! What is the recommendation to avoid infinite storage growth in my S3 bucket that holds my checkpoints? I want to have a lifecycle policy to clean up old SST files (RocksDB with incremental checkpoints) but I find those lifecycle rules still sometimes delete files that are part of my checkpoint. Meaning some SST files never get re-written for very long periods of time but are still part of new checkpoints. We are trying to set

state.backend.rocksdb.compaction.filter.periodic-compaction-time

, but after some reading it seems this setting still doesn't quite do what we want.

I'm really surprised at the slow movement on https://issues.apache.org/jira/browse/FLINK-38770 This seems like a pretty significant bug, autoscaling with the flink operator basically doesn't work with flink 2.0+. The fix has been in a PR ready for review for a while, how long do fixes for issues like this typically take to get reviewed?

Hi! I have a potential use case for broadcast state in this flink app. And i'm trying to understand some of the nuances on how you handle initial start up of a flink app and non-determinism of when the main data stream vs broadcast stream reaches the broadcast process function. The main keyed data stream has messages that contain an id (pointing to the entity this represents), the current state of the entity, and a data provider field. We key the stream on id, and the stream itself contains messages for a single id that might come from different data providers. We have a config stream that contains the a config item like: id, provider to use fields. and we have a processor that will accept the data stream and config stream (as broadcast state) and depending on the config for that message's id, we'll either discard or publish the message through. one piece i'm struggling to figure out is how to handle the scenario where in the broadcast processor, how do i handle the scenario where the config stream takes significantly longer to start consuming and publishing to the broadcast processor, what do i do with the data stream elements? I don't want to drop messages, and I also don't want to just buffer messages indefinitely in a list. Ideally, i'd love it if i was able to introduce some type of semantics like: • apply backpressure to the data stream / don't process or consume the data stream source until the broadcast stream reaches the processor I can't find resources online for this, so i'm wondering if i'm just thinking about this problem incorrectly

Hi everyone I’m using Apache Flink CDC 3.4.0 (Flink 1.2) with the PostgreSQL source connector to replicate data from PostgreSQL into Apache Iceberg on AWS. I’m trying to use Debezium signaling to trigger ad-hoc/incremental snapshots of specific tables. I configured the source with:

--  java --
'debezium.signal.enabled.channels' = 'source',
'debezium.signal.data.collection' = 'public.debezium_signal'

and created the signal table in Postgres:

--- signal table ---

CREATE TABLE public.debezium_signal (
  id VARCHAR(42) PRIMARY KEY,
  type VARCHAR(32) NOT NULL,
  data VARCHAR(2048)
);

Then I insert signals like:

--- trigger adhoc snapshot --- 
INSERT INTO public.debezium_signal (id, type, data)
VALUES (
  '1',
  'execute-snapshot',
  '{"data-collections": ["public.my_table"], "type": "incremental"}'
);

My understanding is that Flink CDC embeds Debezium and passes

debezium.*

properties through, but Flink CDC 3.x also has its own incremental snapshot framework (split enumerator / chunk snapshotting / checkpoint-aware recovery). So I’m unsure whether Flink CDC 3.4.0 actually supports Debezium signaling semantics for triggering snapshots, or whether these configs are simply ignored / only partially honored. Has anyone successfully used Debezium signaling with Flink CDC PostgreSQL source in 3.4.x? Specifically: 1. Does Flink CDC read/process the signal table? 2. Can

execute-snapshot

trigger Flink CDC incremental snapshots? 3. Are there recommended alternatives for ad-hoc table re-snapshotting without restarting the job? Any guidance would be greatly appreciated.

@Will Lauer has left the channel

@Rodney Keeling has left the channel

Hello exists a plan to allow the flink-sql-connector-hive compatible with flink 2.0+? I can see that the connector was moved to an external repo as this ticket says: https://issues.apache.org/jira/browse/FLINK-30064 but there is no releases atm, are you deprecating hive connector? I am interested to use hive metastore with fluss + paimon for tiering service, but I am worried if the hive connector will be deprecated or abandoned like rabbitmq or pulsar connectors.

Hello. Does anyone know what is the proper way to set up ForST to actually do a fast savepoint. Currently its at least 30 times slower than rocksdb. Im testing enabling s5cmd fast-duplicate. But still nothing

Hi Team - I am currently working on integrating Flink with Apicurio Schema registry. So is there any documentation that can guide me with the appropriate setup and libraries required to make the integration work? Any suggestions are appreciated.

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!