This patch provide a option to let balancer blocked for $dfs.balancer.getBlocks.interval.millis milliseconds after every getBlocks RPC call.
The attached pictures shows the improvements after I apply this patch to our production cluster around Thursday 15:00.

yunjiong zhao,
If there are blocks to balance, then there will be sufficient delays between successive getBlocks.
In such cases, we do not have to sleep.
It will be better to keep track of the interval between successive getBlocks and sleep only for the required time.
Can you also write a unit test to cover this change ?

Sleeping inside the Synchronized block should be avoided as it will prevent other threads from obtaining the lock while the thread is sleeping.
One tradeoff in sleeping fixed vs variable time is that code gets complicated. Since by default, the delay is not applied, it is okay to sleep for a fixed interval after getBlocks().

Thank you Benoy Antony for your time to review this patch.

Sleeping inside the Synchronized block should be avoided as it will prevent other threads from obtaining the lock while the thread is sleeping.

I did it on purpose for sleeping inside the Synchronized block.
In balancer there are multiple threads (by default 200) that may call getBlocks at same time, if user need to set dfs.balancer.getBlocks.interval.millis to slow down balancer, without a lock it won't work well due to at worst case there are still 200 getBlocks send to NameNode at same time.

It will be better to keep track of the interval between successive getBlocks and sleep only for the required time.

Since by default, this patch doesn't change anything, only add a option let user slow down balancer send getBlocks to NameNode, so I'd like to keep it as simple as possible.

Thanks for the explanation, yunjiong zhao.
The patch looks good. I will commit this tomorrow if there are no other comments.

-1 overall

This message was automatically generated.

yunjiong zhao, Could you please check to see if sleeping outside the sync block with a potentially longer delay is sufficient to avoid spike ?

Use Semaphore instead lock to avoid findbug warning.

-1 overall

This message was automatically generated.

Hey yunjiong zhao, Benoy Antony. This seems like an important issue to fix. Thanks for reporting and working on the fix. We face the same problem on our cluster.
If I understand correctly there are two changes in the patch:

1. Allow only one Balancer thread to issue getBlocks() at any given time.
2. Add a possibility to sleep for a thread between getBlocks() calls.

Some thoughts

1. I agree with Benoy Antony that the sleep() solution is not optimal. Do you guys think we can replace it with a wait()?
2. I also want to understand what is the effect of this change on the case when there is a lot unbalanced nodes and the Balancer should run aggressively. Can we add some heuristics so that the Balancer could adjust by itself instead of adding the configuration parameter, which would require Balancer restart if it needs to be changed.

Thanks Konstantin Shvachko for review.
Only when you set dfs.balancer.getBlocks.interval.millis to non-zero, Balancer will only allow one thread to issue getBlocks()at any given time. Otherwise this patch doesn't change anything.
So only one change actually.

If use wait, it will release the lock, so can't make sure there are only one thread will call getBlocks().

By default, this patch doesn't change anything. So if you need run Balancer aggressively, don't set dfs.balancer.getBlocks.interval.millis.

Can we add some heuristics so that the Balancer could adjust by itself instead of adding the configuration parameter

I though this before. The best way I can thought is add new function in IPC that let clients get the CallQueueLength, if CallQueueLength is too high, block getBlocks() until the CallQueueLength become normal again.

Took me some time to refresh my memory of the details of balancing. So here is my understanding of what is happening:

• dispatchBlockMoves() spawns a thread for each Source, which represents all storages of the same type of a DN. Each thread executes dispatchBlocks() then.
• dispatchBlocks() first tries to schedule block transfers for already selected (source:target) DN pairs, and if there are no more pairs it calls getBlockList(), which contacts NN to obtain the next portion of blocks from the source DN to be moved out.
• The problem of two many RPC calls happens in the beginning of the Balancer iteration, when there are no scheduled pairs yet, so all the threads call getBlockList() and go to the NameNode simultaneously. So we need to disperse only the initial burst of RPCs at the start of an iteration, as subsequent getBlocks() are already dispersed fine.

I see two ways to fix this:

1. Add a parameter to getBlockList(long delay), where delay is a random time within a reasonable interval, which Balancer should wait for before sending the getBlocks() RPC to NN. The delay is only applied once, and set to 0 once applied. This looks rather straightforward to me.
2. Allocate a reasonable throughput of getBlocks() RPCs to NN, and delay calls if the quota is exceeded. This is similar to Benoy Antony's proposal, but allows to precisely specify how much of NN RPC bandwidth is allocated for the Balancer.

yunjiong zhao I understand you wanted a simple fix without making too much changes, but this looks like a real problem to me, and we should fix it in a more generic manner. I am fine if you wish to implement option #1 here as an initial step. Ultimately we should target solution #2, which could be done in another jira.

Two other approaches to fix this:

1. In getBlockList() Dispatcher fetches the blocks belonging to a particular DN from the NN. And then it moves those blocks from the source DN to the target DN. Dispatcher can instead get the blocks directly from the particular DN. This makes getBlocksList() a distributed operation and doesn't impact any specific node.

2. Dispatcher can fetch the blocks from the Standby NN instead of the active. Balancer should be able to tolerate reasonable degree of staleness.

Great discussions Konstantin Shvachko and Vinitha Reddy Gankidi.

I suggest we keep this JIRA for making the pattern of getBlocks calls less bursty, and also create a JIRA to offload getBlocks calls to DN or SbNN. Vinitha Reddy Gankidi Thoughts?

If we were to offload the calls to DN, dispersing calls wouldn't be a pressing issue. I would like to get some feedback on the various approaches discussed. Benoy Antony, Daryn Sharp, Mingliang Liu and yunjiong zhao I would love to hear your opinions.

Also pinging Ming Ma and Andrew Wang since this is related to "reading from SbNN".

Ping Tsz Wo Nicholas Sze.

Combining my and Vinitha's proposals in a common list here, for further discussion.

1. Add a parameter to getBlockList(long delay), where delay is a random time within a reasonable interval, which Balancer should wait for before sending the getBlocks() RPC to NN. The delay is only applied once, and set to 0 once applied. This looks rather straightforward to me.
2. Allocate a reasonable throughput of getBlocks() RPCs to NN, and delay calls if the quota is exceeded. This is similar to Benoy Antony's proposal, but allows to precisely specify how much of NN RPC bandwidth is allocated for the Balancer.
3. Dispatcher can get the blocks directly from the particular DN instead of NN. This makes getBlocksList() a distributed operation and doesn't impact any specific node.
4. Dispatcher can fetch the blocks from the Standby NN instead of the active. Balancer should be able to tolerate reasonable degree of staleness.

I think (1) is quite easy and can solve immediate problem with bursting RPCs.
I really like Vinitha's (3) obtaining block info directly from DNs. I did not look how much harder it is than other approaches.
(2) and (4) are complimentary, that is we can have a limit on Balancer RPC bandwidth to NN, but run it against SBN.

Konstantin Shvachko I'm leaning towards (4) instead of (3).
isGoodBlockCandidate needs a global view of the block replicas. Also there is some additional logic to deal with erasure coded(EC) blocks and this may be a blocker for reading from DNs. Zhe Zhang you probably have more context regarding the EC blocks.

 /**
   * Decide if the block/blockGroup is a good candidate to be moved from source
   * to target. A block is a good candidate if
   * 1. the block is not in the process of being moved/has not been moved;
   * 2. the block does not have a replica/internalBlock on the target;
   * 3. doing the move does not reduce the number of racks that the block has
   */
  private boolean isGoodBlockCandidate(StorageGroup source, StorageGroup target,
      StorageType targetStorageType, DBlock block) {

I agree that (2) and (4) are complimentary.

Yup, I was thinking the same last night. You need all locations of the block to select the right target that satisfies replication policy, cannot get it from a DN. So unfortunately we should discard (3) as an option.

Here is a relatively simple patch, which restricts the number of RPC calls from Balancer to NN to 20 calls per second.
20 calls per second is a constant for now. It is chosen so that Balancer calls could not saturate NN's RPC queue based on metrics from a large cluster I was observing. LMK if people prefer it to be configurable.
On a large cluster with 200 (default) dispatcher threads, and e.g. 500 underutilized DNs (sources) the initial 200 RPCs will be dispersed over 200 / 20 = 10 seconds. The remaining 300 RPCs should disperse organically as they subsequently reuse the same 200 threads from the pool.
The patch has a unit test, which triggers the dispersion logic.

-1 overall

This message was automatically generated.

-1 overall

This message was automatically generated.

Konstantin Shvachko The delay logic looks good to me. It would be great if we can make BALANCER_NUM_RPC_PER_SEC configurable with a default value of 20.The test does not ensure that there are indeed 20 getBlocks calls per second and it probably is not straightforward to ensure that. So I would like to have the ability to configure BALANCER_NUM_RPC_PER_SEC.

Thanks Konstantin Shvachko for the patch. In addition to Vinitha Reddy Gankidi's suggestion, two minor comments:

1. Typo "Inteval"
2. I guess Preconditions is the preferred way for cases like below:

   assert concurrentThreads > 0 : "Number of concurrent threads is 0.";
   

• I am usually very conservative about introducing new configuration parameters. Parameters seem to give you flexibility to adjust them, but in many cases administrators don't know what to do with that flexibility, because there so many of them. I prefer to have a reasonable constant value initially, and add a config variable later if other value are needed in certain cases. In the end adding configs is easy, but you can never remove them.
  In this particular case the BALANCER_NUM_RPC_PER_SEC is chosen so that big clusters would distribute initial RPC requests over 10 secs, and it does not effect small clusters at all. I think we are good with the constant set to 20 for now, but let me know if you see use cases for different values.
• Fixed the typo in 004 patch. Thanks Zhe Zhang.
• This would be a typical misuse of Preconditions, as we do in many cases in the code, and as it was discussed previously on many occasions. It is an assert, because we assume the condition should never happen. If it does, it's a bug, which should be caught during testing, with -ea option. And in the runtime we want to avoid checking any extra condition for performance reasons.

-1 overall

This message was automatically generated.

Addressing the checkstyle warning.
Not seeing failures of TestBalancer.testBalancerWithStripedFile locally. Don't think it is related.

-1 overall

This message was automatically generated.

Thanks Konstantin Shvachko, the main logic LGTM. I could not reproduce reported unit test failures either.

+1 pending a few final comments:

1. IIUC, BALANCER_NUM_RPC_PER_SEC is a best-effort throttling target, instead of a guaranteed threshold. E.g. it looks possible for Thread.sleep(delay) to be interrupted and getBlockList to be retried in the while loop. Or the entire dispatchBlocks call in a thread could die before delay seconds, then another future[j] will be issued without the delay. (Assuming this understanding is correct), I think this is the right way to handle this logic – it is a good idea not to optimize for these rare cases. But can we update the documentation for BALANCER_NUM_RPC_PER_SEC to reflect it?
2. private void dispatchBlocks(long delay) { doesn't explain delay in its Javadoc.
3. What does testBalancerRPCDelay verify? It is not checking the number of RPC calls.

• You are right, the rate of getBlocks RPCs is not guaranteed. Balancer can only do its best. The actual rate can be only guaranteed on the NameNode, but we don't want to go there.
  I made it clear in the comment for BALANCER_NUM_RPC_PER_SEC.
• Added a decryption for delay.
• It is pretty hard to measure the rate of operations on NN. Here is what I did.
  Created a spy FSNamesystem. The spy would call a modified getBlocks() when the corresponding RPC is called.
  The modified getBlocks() first calls the original method, then counts the number of calls and the time of the first and the last call to getBlocks(). Given the number of calls and the interval we can estimate the rate later on.

-1 overall

This message was automatically generated.

Addressed warnings

-1 overall

This message was automatically generated.

Thanks Konstantin Shvachko for the update. The added Javadoc looks good.

The unit test testBalancerRPCDelay fails for me locally with a NullPointerException.

1. Took some time to reproduce failures. I did not have any on my local box.
   Looks like the solution is to mock FSNamesystem before starting DataNodes. Otherwise the behavior is non-deterministic. I changed it and now it runs consistently on my local box. Let's try Jenkins.
2. findbugs warnings are not related to the patch.
3. There are 2 checkstyle warnings.
   • One complains that the number of parameters in doTest() is more than 7. Don't know why the magical number, but there was 8 parameters in doTest() already and I added one.
   • Second is about inner assignment, which is intentional in this case, because I want the two variables initially have the same value, and splitting the line into two statements would remove that meaning.

Thanks for the update Konstantin Shvachko. Now all other tests in TestBalancer pass except for testBalancerRPCDelay:

java.util.concurrent.TimeoutException: Timed out waiting for /tmp.txt to reach 40 replicas

	at org.apache.hadoop.hdfs.DFSTestUtil.waitReplication(DFSTestUtil.java:764)
	at org.apache.hadoop.hdfs.server.balancer.TestBalancer.createFile(TestBalancer.java:306)
	at org.apache.hadoop.hdfs.server.balancer.TestBalancer.doTest(TestBalancer.java:847)
	at org.apache.hadoop.hdfs.server.balancer.TestBalancer.testBalancerRPCDelay(TestBalancer.java:2071)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:497)
	at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:47)
	at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
	at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:44)
	at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
	at org.junit.internal.runners.statements.FailOnTimeout$StatementThread.run(FailOnTimeout.java:74)

Added waitActive() after starting DataNodes.

-1 overall

This message was automatically generated.

TestBalancerWithSaslDataTransfer fails on my local box too (it passes without the patch).

• New version. It fixes TestBalancerWithSaslDataTransfer.
• Also separated TestBalancerRPCDelay from TestBalancer, since the latter is running more than 5 minutes already.
• The TestDataNodeVolumeFailure series seems to have a long history of failures, looking at HDFS-11398. It fails intermittently with and without my patch.

Thanks Konstantin Shvachko. Latest patch LGTM. +1 pending Jenkins.

-1 overall

This message was automatically generated.

Latest patch added Javadoc to TestBalancerRPCDelay, and corrected one long line.
Merge to branch-2 is pretty straightforward.
Both branch-2.8 and branch-2.7 don't have method getBlocks() in FSNamesystem, so spying on it won't help. I removed mocking for these two branches. Attaching patches for both.

-1 overall

This message was automatically generated.

SUCCESS: Integrated in Jenkins build Hadoop-trunk-Commit #11637 (See https://builds.apache.org/job/Hadoop-trunk-Commit/11637/)
HDFS-11384. Balancer disperses getBlocks calls to avoid NameNode's rpc (shv: rev 28eb2aabebd15c15a357d86e23ca407d3c85211c)

• (edit) hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManagerTestUtil.java
• (edit) hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/NameNodeAdapter.java
• (edit) hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/balancer/TestBalancer.java
• (add) hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/balancer/TestBalancerRPCDelay.java
• (edit) hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java

-1 overall

This message was automatically generated.

Just committed this to 4 branches.
Verified that failed tests for branch-2.7 are passing locally.

Konstantin Shvachko Thanks for the fix.