HBASE-29796 Allow sleepForRetry replication config to be overridden by replication peers #7577
Conversation
This squashed commit combines 8 commits: - Allow peers to override sleep config - Dynamic config update - Always get value - Use protobuf instead of string - Add to test - Add shell command - Use builder instead - Update UI to include sleep
The previous commit incorrectly added methods (getStartPosition, getRecoveredQueueStartPos, terminate) that don't exist in upstream master. These were from the old branch base and should not be included.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
This comment has been minimized.
|
Since we have a Configuration Object in ReplicationPeerConfig, what about just create a combined Configuration instance and use it when creating ReplicationSource? In this way you can directly change the sleepForRetry through the configuration. |
This comment has been minimized.
This comment has been minimized.
Just to confirm, you're suggesting I use the existing Configuration combinedConf = new Configuration(globalConf);
// Override any values in globalConf with the existing value in peerConfig
peerConfig.getConfiguration().forEach(combinedConf::set);
// set this.conf = combinedConf in the ReplicationSourceSet peer-specific override via Is that the approach you are looking for? |
Yes. More specific, we can change the code in ReplicationSourceManager.createSource. We have a CompoundConfiguration in hbase, where we can merge multiple Configurations together. You can check ReplicationPeerConfigUtil.getPeerClusterConfiguration method to find the usage. Thanks. |
There was a problem hiding this comment.
Pull request overview
This pull request adds support for configuring the replication source sleepForRetries parameter on a per-peer basis, with fallback to the global configuration when not set. Previously, this parameter was only configurable globally via the replication.source.sleepforretries property.
- Adds
sleepForRetriesfield toReplicationPeerConfigwith builder support and protobuf serialization - Implements
getSleepForRetries()method inReplicationSourcewith fallback logic to global config when peer value is 0 - Adds shell command
set_peer_sleep_for_retriesfor managing the per-peer configuration
Reviewed changes
Copilot reviewed 16 out of 16 changed files in this pull request and generated 5 comments.
Show a summary per file
| File | Description |
|---|---|
| hbase-protocol-shaded/src/main/protobuf/server/master/Replication.proto | Adds sleep_for_retries field to ReplicationPeer protobuf message |
| hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeerConfig.java | Adds sleepForRetries field to data model with getter/setter and includes in toString() |
| hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeerConfigBuilder.java | Adds setSleepForRetries() method to builder interface |
| hbase-client/src/main/java/org/apache/hadoop/hbase/client/replication/ReplicationPeerConfigUtil.java | Adds conversion logic for sleepForRetries between protobuf and Java objects |
| hbase-client/src/test/java/org/apache/hadoop/hbase/replication/ReplicationPeerConfigTestUtil.java | Updates test utilities to include sleepForRetries in config generation and assertions |
| hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java | Implements getSleepForRetries() with fallback logic and updates all usage of sleepForRetries to call this method |
| hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceInterface.java | Adds getSleepForRetries() method to interface |
| hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALReader.java | Updates to use source.getSleepForRetries() instead of local field |
| hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceShipper.java | Updates to use source.getSleepForRetries() instead of local field |
| hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java | Updates to use source.getSleepForRetries() instead of local field |
| hbase-server/src/test/java/org/apache/hadoop/hbase/replication/ReplicationSourceDummy.java | Implements getSleepForRetries() in test dummy class |
| hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSource.java | Updates mocks to return 0L for getSleepForRetries() |
| hbase-shell/src/main/ruby/shell/commands/set_peer_sleep_for_retries.rb | Adds new shell command for setting per-peer sleep for retries |
| hbase-shell/src/main/ruby/hbase/replication_admin.rb | Adds set_peer_sleep_for_retries() method implementation |
| hbase-shell/src/main/ruby/shell.rb | Registers new set_peer_sleep_for_retries command |
| hbase-shell/src/test/ruby/hbase/replication_admin_test.rb | Adds test for set_peer_sleep_for_retries command |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| public boolean isSerial() { | ||
| return serial; | ||
| } | ||
|
|
There was a problem hiding this comment.
The public getter method getSleepForRetries() is missing Javadoc documentation. For consistency with other public methods in this class and to help users understand the purpose and behavior of this configuration property, add Javadoc that explains:
- What this value represents (sleep time between retries in milliseconds)
- The default value (0, which means use global configuration)
- When it's used (during replication retries)
| /** | |
| * Returns the per-peer sleep time between replication retries, in milliseconds. | |
| * <p> | |
| * A value of {@code 0} means that this peer does not define its own retry sleep and the | |
| * global replication configuration should be used instead. | |
| * This value is consulted when performing replication retries for this peer. | |
| * | |
| * @return sleep time between replication retries in milliseconds, or {@code 0} to use the | |
| * global configuration | |
| */ |
| public long getSleepForRetries() { | ||
| long peerSleepForRetries = replicationPeer.getPeerConfig().getSleepForRetries(); | ||
| if (peerSleepForRetries > 0) { | ||
| return peerSleepForRetries; | ||
| } | ||
| return this.conf.getLong("replication.source.sleepforretries", 1000); | ||
| } |
There was a problem hiding this comment.
The fallback logic is inconsistent with the existing getCurrentBandwidth() pattern. The bandwidth implementation (lines 519-523) treats 0 as "use default", but this implementation treats values > 0 as valid peer settings and 0 as "use default". This creates an inconsistency where setting sleepForRetries to 0 means "use global default" but for bandwidth, 0 means "use default". Consider using the same pattern as bandwidth (checking != 0 instead of > 0) for consistency, or add explicit documentation about this behavior difference.
| Examples: | ||
| # set sleep time to 2 seconds (2000ms) between retries for a peer | ||
| hbase> set_peer_sleep_for_retries '1', 2000 | ||
| # unset sleep time for a peer to use the global default configured in server-side |
There was a problem hiding this comment.
The comment states "unset sleep time for a peer to use the global default configured in server-side" when passing 0, but this behavior differs from the bandwidth configuration pattern shown in the same file. In the bandwidth test (lines 609-622), 0 is also used as the initial/default value, but the bandwidth implementation treats 0 as "use default bandwidth". For consistency and clarity, consider documenting this explicitly in the help text that 0 has special meaning as a sentinel value for "use global default", and ensure users understand that setting it to 0 is not the same as setting it to the actual configured global default value.
| Examples: | |
| # set sleep time to 2 seconds (2000ms) between retries for a peer | |
| hbase> set_peer_sleep_for_retries '1', 2000 | |
| # unset sleep time for a peer to use the global default configured in server-side | |
| A value of 0 is treated specially as a sentinel meaning "use the global default | |
| sleep time configured on the server side" rather than an explicit sleep time. | |
| Examples: | |
| # set sleep time to 2 seconds (2000ms) between retries for a peer | |
| hbase> set_peer_sleep_for_retries '1', 2000 | |
| # unset the peer-specific sleep time so that this peer uses the global default | |
| # configured on the server side (0 is a sentinel and does not change that default) |
| public long getSleepForRetries() { | ||
| long peerSleepForRetries = replicationPeer.getPeerConfig().getSleepForRetries(); | ||
| if (peerSleepForRetries > 0) { | ||
| return peerSleepForRetries; | ||
| } | ||
| return this.conf.getLong("replication.source.sleepforretries", 1000); | ||
| } |
There was a problem hiding this comment.
Missing test coverage for the new getSleepForRetries() method and its fallback logic. While the existing tests mock getSleepForRetries() to return 0L, there's no test that verifies:
- When peer config has a positive value (e.g., 2000), it returns that value
- When peer config is 0, it falls back to the global configuration value
- The integration between peer-level and global configuration
Consider adding a dedicated unit test in TestReplicationSource.java that verifies this fallback behavior, similar to how bandwidth is tested in other parts of the codebase.
| * Get the sleep time for retries. Check peer config first, if set use it, otherwise fall back to | ||
| * global configuration. | ||
| * @return sleep time in milliseconds | ||
| */ |
There was a problem hiding this comment.
This method overrides ReplicationSourceInterface.getSleepForRetries; it is advisable to add an Override annotation.
| */ | |
| */ | |
| @Override |
| this.conf.getInt(WAIT_ON_ENDPOINT_SECONDS, DEFAULT_WAIT_ON_ENDPOINT_SECONDS); | ||
| decorateConf(); | ||
| // 1 second | ||
| this.sleepForRetries = this.conf.getLong("replication.source.sleepforretries", 1000); |
There was a problem hiding this comment.
nit: do you think this sleepForRetries would be changing dynamically after the initialization ? IIRC it would be only reloading via refreshSources (e.g. via updateReplicationPeerConfig ) such that this value would only created once per peer configuration or refresh.
so, can we keep this variable and just call getSleepForRetries() once within ReplicationSource.java?
This comment has been minimized.
This comment has been minimized.
...test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSourceManager.java
Show resolved
Hide resolved
...server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
Show resolved
Hide resolved
| // Check if any replication.source.* keys have changed values | ||
| for (Map.Entry<String, String> entry : newReplicationConfigs.entrySet()) { | ||
| String key = entry.getKey(); | ||
| if (key.startsWith("replication.source.")) { |
There was a problem hiding this comment.
do we need to handle the following keys? if not, then your change looks good to me
hbase.replication.source.fs.conf.provider
hbase.replication.source.service
hbase.replication.source.maxthreads
There was a problem hiding this comment.
As far as I can tell, the first two couldn't be used on a per-peer basis while the last is only used by the InterClusterReplicationEndpoint and not a ReplicationSource. Therefore, I think it is fine to not include those.
|
I have updated #7578 to reflect the changes suggested by PR comments as well. It only has minor differences from this branch. |
This comment has been minimized.
This comment has been minimized.
|
🎊 +1 overall
This message was automatically generated. |
|
💔 -1 overall
This message was automatically generated. |
|
🎊 +1 overall
This message was automatically generated. |
4 participants
Currently, sleepForRetries (the sleep time between retry attempts during replication) is only configurable globally via the replication.source.sleepforretries configuration property. This makes it impossible to tune behavior for individual replication peers that may have different requirements.
This change would add support for overriding replication source config values on a per-peer basis, with fallback to the global configuration when not set.
This is related to #7578 because this will not cleanly merge into branch-2