Balancer can work incorrectly in case of slow historicals or large number of segments to move

[a2l007]

When a cluster has thousands of segments to load, it is possible that historicals take more time to process all the requests showing up in the loadQueue. Consequently, the coordinator timeout druid.coordinator.load.timeout expires and after that the segment request is removed from the CuratorLoadQueuePeon object for this historical and the balancer attempts to balance this segment again.
However, since the zookeeper path for this segment still exists on the earlier historical, this historical will eventually load this segment even though the coordinator has decided to place the segment elsewhere.
This behavior can cause the balancer to make incorrect segment placement decisions since ServerHolder.getSizeUsed() will now be different from the actual used size on that historical. This is more evident when using the disk Normalized balancer strategy as the usageRatio plays an important role in that strategy.

Couple of ways to mitigate this problem would be to increase the druid.coordinator.load.timeout property or lower the maxSegmentsInNodeLoadingQueue so that the historicals aren't flooded with loadqueue entries when there are a large number of segments to move.

Investigating into this issue, it looks like CuratorLoadQueuePeon shouldn't be deleting the load/queue entries in the event of a load timeout, so that the balancer is aware of this activity.

[jihoonson]

@a2l007 thanks for filing this issue. Does this bug make the coordinator confused until the historical announces those segments? Like, if a segment was removed from a loadQueuePeon but a historical load and announced it later, the server view would be updated properly and the coordinator could make a valid decision? I guess it could get worse and worse if there are lots of such segments though. To make the historical process them faster, does setting druid.segmentCache.numLoadingThreads to something high also help?

[jihoonson]

For http segment loading, I think it's fine even though there is a little chance that a similar issue can happen. Since HttpClient in HttpLoadQueuePeon and SegmentListerResource use the same timeout, it's possible that the httpClient can time out even when SegmentListerResource returns successfully. However, even in that case, the historical will announce the segment properly which in turn will update the coordinator server view. @himanshug is this correct?

[a2l007]

@a2l007 thanks for filing this issue. Does this bug make the coordinator confused until the historical announces those segments? Like, if a segment was removed from a loadQueuePeon but a historical load and announced it later, the server view would be updated properly and the coordinator could make a valid decision?

Yeah this is exactly what's happening. Due to this behavior, Coordinator would think a specific set of historicals have enough capacity and would continue to try to assign load requests to it. The problem keeps cascading as there are more segments to load. One of our clusters encountered this issue where a few of the historicals were loaded upto 100% even though there were many other historicals with just 65% filled capacity.

To make the historical process them faster, does setting druid.segmentCache.numLoadingThreads to something high also help?

Yes, that does help to a certain extent as well. However certain clusters tend to set the default value for this property in order to minimize zookeeper flakiness that sometimes accompanies higher druid.segmentCache.numLoadingThreads values.

In terms of a fix for this issue, I think using the failedAssignCount from LoadQueuePeon or something similar as a factor in the balancing strategy could help.

[jihoonson]

Thanks for confirming the issue.

Yes, that does help to a certain extent as well. However certain clusters tend to set the default value for this property in order to minimize zookeeper flakiness that sometimes accompanies higher druid.segmentCache.numLoadingThreads values.

Are there any flakiness problems you are aware of? Maybe we should fix them.

In terms of a fix for this issue, I think using the failedAssignCount from LoadQueuePeon or something similar as a factor in the balancing strategy could help.

Yeah, maybe the assign timeout should be used only for triggering reassign so that segments can be loaded as soon as possible, but the balancer should treat as if those segments were still in the loadQueuePeon.

[a2l007]

Are there any flakiness problems you are aware of? Maybe we should fix them.

It's more of a zookeeper issue than a druid issue I think. Increase in historical zookeeper interactions sometimes lead to weird behavior such as connection timeouts or zookeeper closing the socket connection. This isn't easily reproducible in order to understand what exactly causes this.

Yeah, maybe the assign timeout should be used only for triggering reassign so that segments can be loaded as soon as possible, but the balancer should treat as if those segments were still in the loadQueuePeon.

Let me try out a couple of approaches to fixing this. Will raise a PR once it is tested.

[jihoonson]

Thanks!

[a2l007]

@jihoonson I have a PR out for this. Could you please review it?

[himanshug]

Technically possible even in http if historicals were super slow in downloading segments from deep storage but it is much less likely to happen because all segments to load are not immediately placed in historical's queue of segments to load unlike the zk version where historical's queue is in the zk. Also, there are no problems in case of http when increasing the numLoadingThreads to improve historical's speed of loading segments to match the rate at which they are assigned to it.

from a quick skim over the discussion, solution proposed in #10213 is ok if only one historical was slow/mis-behaving due to a host issue but if this is a cluster level problem then you might not get desired behavior, that solution would end up over allocating segment assignment to historicals, consequently requiring overall "historical cluster" to do more work than what was being done otherwise probably.
I would try to increase numLoadingThreads to improve historical's segment loading speed(.. using http segment loading if you get zk troubles in doing that) and/or adding more historicals if possible.

[clintropolis]

This timeout and 'forgetting' about a load causes another issue coordinator side, where if the coordinator decides to try loading the segment on the same host again after the initial timeout, these type of errors begin appear:

2021-01-30T01:48:03,421 WARN [Master-PeonExec--0] org.apache.druid.server.coordinator.CuratorLoadQueuePeon - ZK node already exists because segment change request hasn't yet been processed
org.apache.zookeeper.KeeperException$NodeExistsException: KeeperErrorCode = NodeExists for /druid/loadQueue/localhost:8083/wiki_2016-06-27T00:00:00.000Z_2016-06-28T00:00:00.000Z_2021-01-26T14:05:06.437Z
	at org.apache.zookeeper.KeeperException.create(KeeperException.java:122) ~[zookeeper-3.4.14.jar:3.4.14-4c25d480e66aadd371de8bd2fd8da255ac140bcf]
	at org.apache.zookeeper.KeeperException.create(KeeperException.java:54) ~[zookeeper-3.4.14.jar:3.4.14-4c25d480e66aadd371de8bd2fd8da255ac140bcf]
	at org.apache.zookeeper.ZooKeeper.create(ZooKeeper.java:792) ~[zookeeper-3.4.14.jar:3.4.14-4c25d480e66aadd371de8bd2fd8da255ac140bcf]
...

until the historical eventually loads the segment (because the now phantom load order still exists) and deletes the load key, clearing the error.

[clintropolis]

from a quick skim over the discussion, solution proposed in #10213 is ok if only one historical was slow/mis-behaving due to a host issue but if this is a cluster level problem then you might not get desired behavior, that solution would end up over allocating segment assignment to historicals, consequently requiring overall "historical cluster" to do more work than what was being done otherwise probably.

@himanshug is it the balancer change in that PR that has you worried I guess that it will cause extra data to be pulled from deep storage? I think that if this change were put in place, what you describe could still be mitigated by increasing the timeout or number of threads, but it doesn't seem to terrible for druid to be able to assume the worst and try to have something else load it, and the coordinator won't make false assumptions about historicals maybe having less space than they actual have due to the now phantom loads since the segments are still tracked on the peon.