[SUPPORT] Hudi Job fails fast in concurrent write even with high retries and long wait time

[Jason-liujc]

Describe the problem you faced

We have a usecase where might have a lot of concurrent writes to the same partition under special scenarios. We are testing if Hudi supports this natively by changing some of the lock retry/wait-time parameters.
We are trying allow all these writers to go through with optimistic retries eventually by using really high num_retries and wait_time_ms parameters.

To Reproduce

We are using DynamoDB as the lock provider, running these loads on AWS EMR

We have the following using options related to concurrency control:

      "hoodie.write.concurrency.mode" -> "optimistic_concurrency_control",
      "hoodie.write.lock.client.wait_time_ms_between_retry" -> "60000",
      "hoodie.write.lock.max_wait_time_ms_between_retry" -> "600000",
      "hoodie.write.lock.num_retries" -> "60",
      "hoodie.write.lock.wait_time_ms" -> "360000",
      "hoodie.write.lock.wait_time_ms_between_retry" -> "360000",
      "hoodie.cleaner.policy.failed.writes" -> "LAZY",
      "hoodie.write.lock.dynamodb.endpoint_url" -> "dynamodb.us-east-1.amazonaws.com",
      "hoodie.write.lock.dynamodb.partition_key" -> "PrepareMergeJob-capitalcasetable-NA",
      "hoodie.write.lock.dynamodb.table" -> "DatastoreWriteLockTable",
      "hoodie.write.lock.provider" -> "org.apache.hudi.aws.transaction.lock.DynamoDBBasedLockProvider",
      "hoodie.write.lock.dynamodb.region" -> "us-east-1",

We spinned up 7 jobs that writes to the same table. Each job should take around ~20 minutes to finish on its own.

Expected behavior

These 7 jobs will have conflicting writes and will retry and will succeed eventually.

Base on the retry parameters I have set, I'd expect it to run for at least 4 hours.

Environment Description

• Hudi version : 0.13.0

• Spark version : 3.3

• Hive version : 3.1.3

• Hadoop version : 3.3.3

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

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

Additional context

Running these workloads on EMR. This is a follow up to this issue: #9512

Stacktrace

Seeing these errors after 0.5 hours for 40% of the jobs:

java.util.ConcurrentModificationException: Cannot resolve conflicts for overlapping writes
	at org.apache.hudi.client.transaction.SimpleConcurrentFileWritesConflictResolutionStrategy.resolveConflict(SimpleConcurrentFileWritesConflictResolutionStrategy.java:108)
	at org.apache.hudi.client.utils.TransactionUtils.lambda$resolveWriteConflictIfAny$0(TransactionUtils.java:85)
	at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1384)
	at java.util.stream.Streams$ConcatSpliterator.forEachRemaining(Streams.java:742)
	at java.util.stream.Streams$ConcatSpliterator.forEachRemaining(Streams.java:742)
	at java.util.stream.ReferencePipeline$Head.forEach(ReferencePipeline.java:647)

[danny0405]

OCC is not designed for high frequency concurrent access, we are working hard towards non-blocking concurrency control: #7907

[Jason-liujc]

Thanks! That's good to know. Meanwhile we can build our own "lock"

However, does this symptom also mean these retry parameters does not work for DynamoDB based lock provider? I was looking into the source code for 0.13.0 and seems DynamoDBLockProvider is not picking up some of the parameters like "write.lock.num_retries". Is that true?

[ad1happy2go]

@Jason-liujc I did a fix around the same line which will be included in 0.14. Dynamodb config class had some configuration issues which was not allowing it to pick default params. Although I haven't tried this around num_retries. Can you try with this PR - #8868

[Jason-liujc]

Thanks. We've tried the newest update from DynamoDBBasedLockProvider and DynamoDbBasedLockConfig but we are still seeing jobs fail pretty soon if encountering conflicting writes.

Just going through the source code, I'm seeing ZookeeperBasedLockProvider is using retry under the hood:

hudi/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/ZookeeperBasedLockProvider.java

Line 58 in 9e9f768

public class ZookeeperBasedLockProvider implements LockProvider<InterProcessMutex>, Serializable {

Whereas DynamoDB is not:

hudi/hudi-aws/src/main/java/org/apache/hudi/aws/transaction/lock/DynamoDBBasedLockProvider.java

Line 62 in 9e9f768

public class DynamoDBBasedLockProvider implements LockProvider<LockItem> {

My theory right now is when we are using DynamoDB Lock Provider, the retry parameters are not used, hence the job fails pretty fast. (i.e. hoodie.write.lock.num_retries). Since going through the Hudi created DynamoDB entry, I don't see any state regarding how many retries a given lock was used.

Let me know if that's the case or an expected behavior from Hudi.

[psendyk]

@Jason-liujc I believe the issue you're seeing is unrelated to lock acquisition. From the multi-writer docs:

For example, when two writers write to non overlapping files, both writes are allowed to succeed. However, when the writes from different writers overlap (touch the same set of files), only one of them will succeed

From the error you posted, it seems like both writers write to the same file ID so one of them fails. The PR that @danny0405 linked addresses this issue by resolving the conflicts during compaction/read-time.

[SamarthRaval]

Thanks. We've tried the newest update from DynamoDBBasedLockProvider and DynamoDbBasedLockConfig but we are still seeing jobs fail pretty soon if encountering conflicting writes.

Just going through the source code, I'm seeing ZookeeperBasedLockProvider is using retry under the hood:

hudi/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/ZookeeperBasedLockProvider.java

Line 58 in 9e9f768

public class ZookeeperBasedLockProvider implements LockProvider<InterProcessMutex>, Serializable {

Whereas DynamoDB is not:

hudi/hudi-aws/src/main/java/org/apache/hudi/aws/transaction/lock/DynamoDBBasedLockProvider.java

Line 62 in 9e9f768

public class DynamoDBBasedLockProvider implements LockProvider<LockItem> {

My theory right now is when we are using DynamoDB Lock Provider, the retry parameters are not used, hence the job fails pretty fast. (i.e. hoodie.write.lock.num_retries). Since going through the Hudi created DynamoDB entry, I don't see any state regarding how many retries a given lock was used.

Let me know if that's the case or an expected behavior from Hudi.

Can we please confirm this issue if it persists ?
Should we start using zookeeper rather then DynamoDB lock ?

[ad1happy2go]

@SamarthRaval @Jason-liujc As discussed, the retry configuration is unrelated to the problem you are facing. The only way to handle such scenario's at this moment will be handling retries at your application level code.

Hopefully #7907 will solve this problem in hudi 1.0