[SUPPORT] Unstable Flink-Hudi connector on scale

[fenil25]

Tips before filing an issue

• Have you gone through our FAQs? Yes

• Join the mailing list to engage in conversations and get faster support at dev-subscribe@hudi.apache.org.

• If you have triaged this as a bug, then file an issue directly.

Describe the problem you faced

Our pipeline involves ingesting the changelog from Kafka into Flink and then we finally use the Hudi sink. We are observing a lot of issues with few tables. The table emits around 10K upserts per second (a huge number of updates). We are using EMR 6.11.0 with Hudi version 13.0 and Flink version 1.16.0 Initially, we tried Copy On Write (COW) but then saw lot of issues with checkpointing in Flink. The main culprit was the error - Checkpoint expired before completing. Increasing resources, decreasing checkpoint interval, increasing checkpoint timeout nothing helped. We then moved to Merge On Read table. We were still seeing issues like - java.lang.IllegalStateException: Receive an unexpected event for instant 20230912181658265 from task 7 Open hudi issue like this suggest that it's multiple writer issue. We set the writers to 1 to resolve this and then increasing checkpoint timeout from 15 minutes to 60 minutes helped. The checkpoint was taking around 30 minutes. However, the main problem we faced here was with whatever config we change for the Flink pipeline, we had to rebootstrap the table. This was not the case with CoW table. Bootstrapping is expensive for us and takes quite some time. If we do not bootstrap, then we see the error of IllegalStateException again or FileAlreadyExistsException (for log files) Our main questions are -

• Is there any way where we can circumvent bootstrapping for flink config changes like changing checkpoint interval etc for the MOR option.
• For such a pipeline, how do we parallelize this with multiple writers without running into aforementioned IllegalStateException/FileAlreadyExistsException problem.
• What better configs should we set to increase the efficiency of the ingestion?

To Reproduce

Steps to reproduce the behavior:

1. On EMR, setup an ingestion from MySQL Debezium changelogs emitted to Kafka and finally consumed by Flink to create a Hudi table
2. The scale of changelogs should be around 10K upserts/sec
3. Hudi/Flink configs
   • MOR table
   • Single writer
   • Checkpoint configuration -
   • hudi.clean.policy=KEEP_LATEST_BY_HOURS
   • --hudi.clean.retain_hours=168
   • All other configs are just default configurations

Environment Description

• Hudi version : 13.0

• Flink version : 1.16.0

• Hive version :

• Hadoop version :

• Storage (HDFS/S3/GCS..) : S3

• Running on Docker? (yes/no) : no

Stacktrace

Stack trace of the aforementioned errors -

• FileAlreadyExistsException

  org.apache.hudi.exception.HoodieRollbackException: Failed to rollback s3://slack-dw/hudi/channels_members commits 20230912181658265 
  ...
  Caused by: org.apache.hudi.exception.HoodieException: Error occurs when executing flatMap
  ...
  Caused by: org.apache.hudi.exception.HoodieRollbackException: Failed to rollback for instant [==>20230912181658265__deltacommit__INFLIGHT]
  ...
  Caused by: org.apache.hadoop.fs.FileAlreadyExistsException: File already exists:s3://slack-dw/hudi/channels_members/.108ba95b-d395-4b98-826d-9e8e0b26d8e1-1_20230912004700323.log.1247_1-0-1
  ...
  2023-09-13 19:50:06,654 INFO  org.apache.flink.runtime.jobmaster.JobMaster                 [] - Trying to recover from a global failure.
  org.apache.flink.util.FlinkException: Global failure triggered by OperatorCoordinator for 'stream_write: channels_members' (operator cd7a24f114e59daa8b573b2d08871155).
  at org.apache.flink.runtime.operators.coordination.OperatorCoordinatorHolder$LazyInitializedCoordinatorContext.failJob(OperatorCoordinatorHolder.java:617) ~[flink-dist-1.16.0.jar:1.16.0] 

• IllegalStateException

  org.apache.flink.util.FlinkException: Global failure triggered by OperatorCoordinator for 'stream_write: channels_members' (operator cd7a24f114e59daa8b573b2d08871155).
  at org.apache.flink.runtime.operators.coordination.OperatorCoordinatorHolder$LazyInitializedCoordinatorContext.failJob(OperatorCoordinatorHolder.java:617)
  at org.apache.hudi.sink.StreamWriteOperatorCoordinator.lambda$start$0(StreamWriteOperatorCoordinator.java:190)
  at org.apache.hudi.sink.utils.NonThrownExecutor.handleException(NonThrownExecutor.java:142)
  at org.apache.hudi.sink.utils.NonThrownExecutor.lambda$wrapAction$0(NonThrownExecutor.java:133)
  at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
  at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
  at java.base/java.lang.Thread.run(Thread.java:829)
  Caused by: org.apache.hudi.exception.HoodieException: Executor executes action [handle write metadata event for instant ] error
  ... 6 more
  Caused by: java.lang.IllegalStateException: Receive an unexpected event for instant 20230912181658265 from task 7
  at org.apache.hudi.common.util.ValidationUtils.checkState(ValidationUtils.java:67)
  at org.apache.hudi.sink.StreamWriteOperatorCoordinator.handleWriteMetaEvent(StreamWriteOperatorCoordinator.java:447)
  at org.apache.hudi.sink.StreamWriteOperatorCoordinator.lambda$handleEventFromOperator$4(StreamWriteOperatorCoordinator.java:288)
  at org.apache.hudi.sink.utils.NonThrownExecutor.lambda$wrapAction$0(NonThrownExecutor.java:130)
  ... 3 more

[danny0405]

Sorry for the issue @fenil25 , the release 0.13.0 is a very buggy release, I'm wondering if you can try rellease 0.12.3 or 0.13.1 instead.

[ad1happy2go]

@fenil25 Were you able to try with the 0.12.3 or 0.13.1. Did you still faced this issue?