Improve performance of zookeeper based mechanism for loading/unloading segments to/from a Druid cluster

[samarthjain]

I have been investigating why clusters have been taking a long time to download segments. After adding a few log lines and closely inspecting the code, it looks like that a zookeeper node for a segment is not created till the processing (download and memory mapping) of previous segment is complete.

See the below code blocks in CuratorLoadQueuePeon

In processSegmentChangeRequest(), we have this piece of block that prevents processing of new segment change requests if there is one in progress.

if (currentlyProcessing != null) {
     log.debug(
          "Server[%s] skipping processSegmentChangeRequest because something is currently loading[%s].",
          basePath,
          currentlyProcessing.getSegmentId()
      );
      return;
 }

In non-failure conditions, currentlyProcessing is set to null only after the zk node for the segment has been deleted

case NodeDeleted:
                log.info("Watcher notified of %s deletion", watchedEvent.getPath());
                entryRemoved(watchedEvent.getPath());
                break;

entryRemoved() calls actionCompleted() which is where we end up setting currentlyProcessing to null

final List<LoadPeonCallback> callbacks = currentlyProcessing.getCallbacks();
      currentlyProcessing = null;

Code in ZkCoordinator which does the node deletion in a callback after the download segment request is complete:

 request.go(
                          dataSegmentChangeHandler,
                          new DataSegmentChangeCallback()
                          {
                            boolean hasRun = false;

                            @Override
                            public void execute()
                            {
                              try {
                                if (!hasRun) {
>>>>>>>                       curator.delete().guaranteed().forPath(path);
                                  hasRun = true;
                                }
                              }
                              catch (Exception e) {
                                try {
                                  curator.delete().guaranteed().forPath(path);
                                }
                                catch (Exception e1) {
                                  log.error(e1, "Failed to delete zNode[%s], but ignoring exception.", path);
                                }
                                log.error(e, "Exception while removing zNode[%s]", path);
                                throw Throwables.propagate(e);
                              }
                            }
                          }
                      

Consequently, majority of historical nodes stay idle for periods doing nothing even though they have the bandwidth to download the segments.

Even after solving the above problem, we still need to improve the concurrency of downloading segments from cold storage.

In ZkCoordinator, a listener is installed on the PathChildrenCache to detect when a node corresponding to the segment to be processed is added. However, downloading and memory mapping of the segment happens in the same thread that is used to handle this event.

This severely limits the throughput at which segments can be downloaded from the cluster. So to improve the performance, my proposal is to improve the parallelism of the two components in the following ways:

1. In CuratorBasedLoadQueuePeon - remove the restriction of only processing one segment at a time. And utilize a thread pool to create zookeeper nodes corresponding to segments that need to be processed (load/drop).
2. In ZkCoordinator, utilize a thread pool to download the segments from the historical. This new threadpool, to do the actual segment processing, will be different from the single threaded pool being used by the PathChildren cache for handling child created events.

[samarthjain]

@fjy or @gianm - would you happen to remember if this was a deliberate design decision? Before I go down the path of making changes, I want to be sure I am not missing out on some tricky issues here.

The issue is fairly severe. In my logs, I see that it took over 5 hours for the coordinator to create zk entries and roughly the same time for historicals to download 3TB of segments from s3.

We have also seen that in some cases, after a largish batch ingestion of a few hundred GBs, the segments are not available on the cluster for almost a couple of hours.

[gianm]

@fjy or @gianm - would you happen to remember if this was a deliberate design decision? Before I go down the path of making changes, I want to be sure I am not missing out on some tricky issues here.

I don't remember that far back - sorry! I would have to guess it was done that way because it was easier and didn't cause a problem at the time. Fwiw, HttpLoadQueuePeon does do things in a more parallel way, but, given that it is still experimental I think there is value in adding better parallelism to the CuratorBasedLoadQueuePeon mechanism as well.

1. In CuratorBasedLoadQueuePeon - remove the restriction of only processing one segment at a time. And utilize a thread pool to create zookeeper nodes corresponding to segments that need to be processed (load/drop).
2. In ZkCoordinator, utilize a thread pool to download the segments from the historical. This new threadpool, to do the actual segment processing, will be different from the single threaded pool being used by the PathChildren cache for handling child created events.

The basic idea sounds good to me. The only potential thing to watch out for, that I can think of, is making sure that code that cleans up directories (e.g. when a segment fails to download, or when the last segment of a particular interval is dropped) doesn't have any races with the code that loads up new segments.

[glasser]

How experimental is the HTTP load queue peon these days? I see that you can turn it on by setting druid.coordinator.loadqueuepeon.type=http and I see the original druid-dev thread by @himanshug but not much discussion afterwards. Are people using it in production?

[pantaoran]

I would love to use HTTP loading but currently I can't because of #5325
This is resolved in OpenSource Druid 0.13.0 but we haven't been able to switch to that version yet. On 0.12.3 it is buggy.
I believe that the fix was backported in the Imply version, but we're not using that.

[gianm]

How experimental is the HTTP load queue peon these days? I see that you can turn it on by setting druid.coordinator.loadqueuepeon.type=http and I see the original druid-dev thread by @himanshug but not much discussion afterwards. Are people using it in production?

I know of a few people using it in production. It seems to work fairly well. I have seen some issues here and there with segment loading seeming to get "stuck" or out of sync, but at the time those issues were reported, I wasn't able to narrow it down to whether the root cause was related to HTTP load queue or HTTP segment announcing (both of which were enabled) or maybe even some third thing.

[samarthjain]

Contributed and merged in #7088

[mustafacse]

@samarthjain is this change merged with 0.19.0, i am not able to see this change there or even in master branch