Flink 2.1: Enhance ZkLockFactory: add maxSleepTimeMs and retryPolicyName to support multiple retry policies. by slfan1989 · Pull Request #14243 · apache/iceberg

slfan1989 · 2025-10-03T03:02:24Z

Summary

This PR enhances ZkLockFactory by introducing two new configuration options:

maxSleepTimeMs
retryPolicyName

Previously, ZkLockFactory only supported the EXPONENTIAL_BACKOFF retry strategy. With this change, users can now configure different retry policies via retryPolicyName (e.g., ONE_TIME, N_TIME, BOUNDED_EXPONENTIAL_BACKOFF, UNTIL_ELAPSED).

Changes

Added maxSleepTimeMs parameter to allow bounded exponential backoff and until-elapsed policies.
Added retryPolicyName parameter to select the retry strategy dynamically.
Refactored retry policy creation logic into a switch/factory method for better extensibility.
Maintained backward compatibility: if no retryPolicyName is provided, it defaults to EXPONENTIAL_BACKOFF.

Motivation

This improves flexibility and robustness of ZkLockFactory by giving users fine-grained control over retry behavior, rather than being limited to a single exponential backoff policy.

slfan1989 · 2025-10-03T03:31:27Z

@pvary @mxm A minor enhancement that improves ZkLockFactory by adding support for all native ZooKeeper retry policies.

Could you please help review this PR? Thank you very much!

pvary · 2025-10-03T05:48:07Z

Cc: @Guosmilesmile

slfan1989 · 2025-10-07T14:05:08Z

@pvary @mxm A minor enhancement that improves ZkLockFactory by adding support for all native ZooKeeper retry policies.

Could you please help review this PR? Thank you very much!

@pvary When you have some time, Could you please help review this PR? Thank you very much!

The main purpose of this change is to enhance the ZooKeeper lock factory (ZkLockFactory) by introducing configurable retry policies.

Here’s a brief summary of the newly added retry mechanisms:

ONE_TIME — Retries only once after the first failure; a simple and fast-failing strategy.
N_TIME — Keeps retrying until the maximum number of retries is reached.
BOUNDED_EXPONENTIAL_BACKOFF — Similar to exponential backoff, but capped by a maximum sleep time (maxSleepTimeMs).
UNTIL_ELAPSED — Keeps retrying until a specified total elapsed time limit is reached, then stops.

These policies make the retry behavior of ZkLockFactory more flexible and adaptable across different deployment environments, while maintaining backward compatibility with the existing implementation.

Guosmilesmile

Thanks for the pr ! Can we also update the corresponding documentation?

Guosmilesmile · 2025-10-09T02:51:13Z

+      policy = ZKRetryPolicies.valueOf(retryPolicyName);
+    } catch (IllegalArgumentException e) {
+      policy = ZKRetryPolicies.EXPONENTIAL_BACKOFF;


Can we add some log output? Swallowing exceptions directly can leave users puzzled.

Thank you very much for reviewing my code! I’ll make improvements based on your suggestions as soon as possible.

Guosmilesmile · 2025-10-09T03:16:21Z

+
+      case EXPONENTIAL_BACKOFF:
+      default:
+        return new ExponentialBackoffRetry(maxSleepTimeMs, maxRetries);


The maxSleepTimeMs here should be baseSleepTimeMs.

Guosmilesmile · 2025-10-09T03:18:44Z

+        return new RetryNTimes(maxRetries, baseSleepTimeMs);
+
+      case BOUNDED_EXPONENTIAL_BACKOFF:
+        return new BoundedExponentialBackoffRetry(baseSleepTimeMs, maxSleepTimeMs, maxRetries);


What would happen if baseSleepTimeMs equals maxSleepTimeMs and maxRetries is greater than 1?

After reviewing the ExponentialBackoffRetry implementation, if baseSleepTimeMs == maxSleepTimeMs and maxRetries > 1, the retry interval calculation

sleepTimeMs = Math.min(maxSleepTimeMs, baseSleepTimeMs * 2^retryCount)

will always result in sleepTimeMs = maxSleepTimeMs.
In this case, the exponential backoff effectively degrades into a fixed-interval retry strategy.

The relevant code is as follows:

https://github.com/apache/curator/blob/3bc3ea1904c62b7e3e0b94a079a13fd88790df1b/curator-client/src/main/java/org/apache/curator/retry/ExponentialBackoffRetry.java#L65-L73

protected long getSleepTimeMs(int retryCount, long elapsedTimeMs) { // copied from Hadoop's RetryPolicies.java long sleepMs = (long) baseSleepTimeMs * Math.max(1, random.nextInt(1 << (retryCount + 1))); if (sleepMs > maxSleepMs) { log.warn("Sleep extension too large ({}). Pinning to {}", sleepMs, maxSleepMs); sleepMs = maxSleepMs; } return sleepMs; }

Thanks , make sense.

Guosmilesmile · 2025-10-09T03:20:30Z

+        1000,
+        3,
+        2000,
+        ZKRetryPolicies.EXPONENTIAL_BACKOFF.name());


Can we add corresponding test cases to ensure that the chosen strategy actually works?

Thank you for the suggestion! I’ve added some unit tests. Since these retry strategies are implemented and maintained by Apache Curator, I think it should be sufficient to verify that the configuration parameters can be correctly parsed and mapped. Does that sound reasonable to you?

slfan1989 · 2025-10-14T08:35:36Z

Thanks for the pr ! Can we also update the corresponding documentation?

Sorry for the late reply! I’ve updated the code according to your suggestions. For the documentation part, I’ve added the new parameters to flink-maintenance.md. If you have some time, I’d really appreciate it if you could take a look. Thanks a lot!

Guosmilesmile · 2025-10-16T08:43:21Z

 | `flink-maintenance.lock.zookeeper.connection-timeout-ms` | Connection timeout (ms) | `15000` |
 | `flink-maintenance.lock.zookeeper.max-retries` | Max retries | `3` |
 | `flink-maintenance.lock.zookeeper.base-sleep-ms` | Base sleep between retries (ms) | `3000` |
+| `flink-maintenance.lock.zookeeper.max-sleep-ms`          | Maximum sleep time (in milliseconds) between retries. Caps the exponential backoff delay.  | `10000` |


Maximum sleep time (ms) between retries.

Thank you for your review and suggestions! I have made improvements based on your feedback.

Guosmilesmile · 2025-10-16T08:45:58Z

+        return new RetryNTimes(maxRetries, baseSleepTimeMs);
+
+      case BOUNDED_EXPONENTIAL_BACKOFF:
+        return new BoundedExponentialBackoffRetry(baseSleepTimeMs, maxSleepTimeMs, maxRetries);


Thanks , make sense.

Guosmilesmile · 2025-10-16T08:48:55Z

+          assertThat(className).isEqualTo("ExponentialBackoffRetry");
+          break;
+        default:
+          throw new IllegalStateException("Unhandled policy type: " + policy);


If we get a error policy name, we will use ExponentialBackoffRetry in default？

Yes, if the policy name is invalid, we’ll use ExponentialBackoffRetry as the default.

Guosmilesmile · 2025-10-16T08:49:50Z

@pvary Peter , If you have time, please take some time to review it , thanks !

pvary · 2025-10-16T10:54:25Z

      LOG.debug("ZkLockFactory already opened for lockId: {}.", lockId);
      return;
    }
-


Nit: unnecessary change

Got it, I’ll revert this change.

Please add back the new line. It is required after every block

pvary · 2025-10-16T10:58:25Z

+
+    ZKRetryPolicies policy;
+    try {
+      policy = ZKRetryPolicies.valueOf(retryPolicyName);


Could we check sooner that the config is valid?

We can have an enum as an input instead of the string

That makes sense. I’ll update the code to validate the config earlier and switch to using an enum instead of a string.

pvary · 2025-10-16T11:00:22Z

+  /** Gets the Zookeeper retry policy configuration. */
+  public String zkRetryPolicy() {
+    return confParser
+        .stringConf()


Could we use enumConfParser?

Sure, that sounds good — I’ll update it to use enumConfParser.

pvary · 2025-10-20T12:28:24Z

  private final int baseSleepTimeMs;
  private final int maxRetries;
-  private final String retryPolicyName;
+  private volatile ZKRetryPolicies retryPolicy;


Why is this volatile?

volatile is unnecessary because retryPolicy is immutable after construction.

There’s no concurrent modification or visibility issue, so final would be more appropriate.

pvary · 2025-10-20T12:38:08Z

  }
+
+  @Test
+  void testAllRetryPoliciesCreationAndType() {


Something like this?

private static Stream<Arguments> retryPolicyProvider() { return Stream.of( Arguments.of(ZKRetryPolicies.ONE_TIME, RetryOneTime.class), Arguments.of(ZKRetryPolicies.N_TIME, RetryNTimes.class), Arguments.of(ZKRetryPolicies.EXPONENTIAL_BACKOFF, ExponentialBackoffRetry.class), Arguments.of( ZKRetryPolicies.BOUNDED_EXPONENTIAL_BACKOFF, BoundedExponentialBackoffRetry.class), Arguments.of(ZKRetryPolicies.UNTIL_ELAPSED, RetryUntilElapsed.class)); } @ParameterizedTest @MethodSource("retryPolicyProvider") void testRetryPolicyCreationAndType( ZKRetryPolicies policy, Class<? extends RetryPolicy> expectedClass) { ZkLockFactory factory = new ZkLockFactory("localhost:2181", "test", 3000, 3000, 1000, 3, 2000, policy); RetryPolicy retryPolicy = factory.createRetryPolicy(); assertThat(retryPolicy) .as("RetryPolicy should not be null for policy %s", policy) .isNotNull() .as("Expected %s for policy %s", expectedClass.getSimpleName(), policy) .isInstanceOf(expectedClass); }

Yes, exactly — that looks good to me.
I’ll update the code accordingly.

pvary · 2025-10-20T14:26:39Z

+  private transient RetryPolicy retryPolicy;
+  private final int maxSleepTimeMs;


nit: Change the order please, so group the final and the transient attributes

Thanks, updated the field order so that all final configuration fields are grouped together, followed by the transient runtime fields.

pvary · 2025-10-20T14:32:49Z

+    if (retryPolicy == null) {
+      LOG.debug(
+          "RetryPolicy was null after deserialization. Recreating from enum: {}", zkRetryPolicy);
+      retryPolicy = createRetryPolicy(zkRetryPolicy);
+    }


This will effectively be always null. And called only once, since the open is called only once.

I think the previous version was better, like:

this.client = CuratorFrameworkFactory.builder() .connectString(connectString) .sessionTimeoutMs(sessionTimeoutMs) .connectionTimeoutMs(connectionTimeoutMs) .retryPolicy(new ExponentialBackoffRetry(baseSleepTimeMs, maxRetries)) .retryPolicy(createPolicy()) .build();

Good point — you’re right, the retryPolicy field isn’t reused and would always be null before open().
I’ll simplify it by removing the field and directly calling createRetryPolicy() in the builder.

pvary · 2025-10-20T14:34:42Z

  }
+
+  @VisibleForTesting
+  RetryPolicy createRetryPolicy(ZKRetryPolicies policy) {


This parametrization is strange.
Either:

Use a static method and add all of the required inputs

Use a non-static method, then don't add parameters which are already available in an attribute

Thanks — agreed, the current method doesn’t need extra parameters since all required values are already stored in the instance.

I’ll keep it as a non-static method that directly uses the existing attributes.

pvary · 2025-10-20T16:19:39Z

+      int maxSleepTimeMs,
+      ZKRetryPolicies retryPolicy) {


not that important, but maybe follow the same order as the attributes, and the value settings? 😄

Thanks, makes sense. I’ve reordered the parameters and assignments to match the field order for better readability.

pvary

Small nits, and then +1
Pending tests

Guosmilesmile

LGTM, wait the ci fail fix.

pvary · 2025-10-21T10:28:01Z

@slfan1989: Do you happen to know why this commit is Unverified?

The description shows this:

Are you the one who set this verification request? Are you ok with me merging this PR?

slfan1989 · 2025-10-21T10:43:16Z

@slfan1989: Do you happen to know why this commit is Unverified?

The description shows this:

Are you the one who set this verification request? Are you ok with me merging this PR?

@pvary Thank you for your question! I believe the submission was marked as unverified because I haven't configured GPG verification after switching to a new computer. You can merge this PR, and I greatly appreciate it. I will complete the GPG setup as soon as possible to ensure future submissions are verified. Thank you for your understanding!

slfan1989 · 2025-10-21T10:58:10Z

I’ll disable Vigilant Mode first to avoid any misunderstandings.

…ame to support multiple retry policies.

slfan1989 · 2025-10-21T11:48:36Z

The local GPG has been configured, I rebased the code and force-pushed it, and we can now see the "verified" badge.

pvary · 2025-10-21T13:06:13Z

Merged to main.
Thanks for the PR @slfan1989 and @Guosmilesmile for the review!

@slfan1989: Please backport the change to the other Flink versions.
This could be done with a command like this:

g diff HEAD^ flink/v2.1 |sed "s/v2.1/v2.0/g">/tmp/patch;g apply -3 -p1 /tmp/patch

Please in the backport PR mention if the backport was clean, or you needed to modify anything, so we can review accordingly.

Thanks!

slfan1989 · 2025-10-21T13:24:42Z

Merged to main. Thanks for the PR @slfan1989 and @Guosmilesmile for the review!

@slfan1989: Please backport the change to the other Flink versions. This could be done with a command like this:
g diff HEAD^ flink/v2.1 |sed "s/v2.1/v2.0/g">/tmp/patch;g apply -3 -p1 /tmp/patch
Please in the backport PR mention if the backport was clean, or you needed to modify anything, so we can review accordingly.

Thanks!

@pvary @Guosmilesmile Thank you very much for helping review the code! I will backport this PR to Flink 2.0 and Flink 1.20.

…ry to support multiple retry policies. (#14389) backports #14243

…port multiple retry policies. (apache#14243)

…ry to support multiple retry policies. (apache#14389) backports apache#14243

…port multiple retry policies. (apache#14243)

…ry to support multiple retry policies. (apache#14389) backports apache#14243

github-actions Bot added the flink label Oct 3, 2025

slfan1989 marked this pull request as ready for review October 3, 2025 03:28

Guosmilesmile reviewed Oct 9, 2025

View reviewed changes

github-actions Bot added the docs label Oct 14, 2025

Guosmilesmile reviewed Oct 16, 2025

View reviewed changes

pvary reviewed Oct 16, 2025

View reviewed changes

github-actions Bot added the INFRA label Oct 20, 2025

pvary reviewed Oct 20, 2025

View reviewed changes

Comment thread flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ZkLockFactory.java

pvary reviewed Oct 20, 2025

View reviewed changes

Comment thread flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ZkLockFactory.java

pvary reviewed Oct 20, 2025

View reviewed changes

pvary approved these changes Oct 20, 2025

View reviewed changes

slfan1989 mentioned this pull request Oct 21, 2025

Hotfix: adapt HMS tests to HiveTableOperations ctor (add KeyManagementClient) #14384

Merged

Guosmilesmile approved these changes Oct 21, 2025

View reviewed changes

slfan1989 force-pushed the zklock-retry-policy-enhancement branch from 8b1ae0f to 398f5a6 Compare October 21, 2025 10:51

github-actions Bot added the hive label Oct 21, 2025

slfan1989 force-pushed the zklock-retry-policy-enhancement branch from cc94322 to 3d6af71 Compare October 21, 2025 10:54

Flink 2.1: Enhance ZkLockFactory: add maxSleepTimeMs and retryPolicyN…

38eb18b

…ame to support multiple retry policies.

slfan1989 force-pushed the zklock-retry-policy-enhancement branch from 3d6af71 to 38eb18b Compare October 21, 2025 11:44

pvary merged commit 17cb11b into apache:main Oct 21, 2025
43 checks passed

slfan1989 mentioned this pull request Oct 21, 2025

Flink: Backport Enhance ZkLockFactory: add maxSleepTimeMs and retryPolicyName to support multiple retry policies. #14389

Merged

pvary pushed a commit that referenced this pull request Oct 22, 2025

Flink: Backport add maxSleepTimeMs and retryPolicyName to ZkLockFacto…

f6632e9

…ry to support multiple retry policies. (#14389) backports #14243

thomaschow pushed a commit to thomaschow/iceberg that referenced this pull request Jan 19, 2026

Flink: Add maxSleepTimeMs and retryPolicyName to ZkLockFactory to sup…

a51c197

…port multiple retry policies. (apache#14243)

thomaschow pushed a commit to thomaschow/iceberg that referenced this pull request Jan 19, 2026

Flink: Backport add maxSleepTimeMs and retryPolicyName to ZkLockFacto…

b1485e3

…ry to support multiple retry policies. (apache#14389) backports apache#14243

talatuyarer pushed a commit to talatuyarer/iceberg that referenced this pull request Apr 1, 2026

Flink: Add maxSleepTimeMs and retryPolicyName to ZkLockFactory to sup…

82e9c89

…port multiple retry policies. (apache#14243)

talatuyarer pushed a commit to talatuyarer/iceberg that referenced this pull request Apr 1, 2026

Flink: Backport add maxSleepTimeMs and retryPolicyName to ZkLockFacto…

31040bb

…ry to support multiple retry policies. (apache#14389) backports apache#14243

		private transient RetryPolicy retryPolicy;
		private final int maxSleepTimeMs;

Conversation

slfan1989 commented Oct 3, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Changes

Motivation

Uh oh!

slfan1989 commented Oct 3, 2025

Uh oh!

pvary commented Oct 3, 2025

Uh oh!

slfan1989 commented Oct 7, 2025

Uh oh!

Guosmilesmile left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

slfan1989 commented Oct 14, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Guosmilesmile commented Oct 16, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

slfan1989 commented Oct 3, 2025 •

edited

Loading