Support multi coordinator while serving resource group info

[abhiseksaikia]

Currently v1/resourceGroupState/* endpoint serves resource group info available within a particular coordinator. The change here is to enable these endpoints to retrieve cluster wide resource group info.

This PR depends on #16163

Test plan - unit test

== RELEASE NOTES ==

General Changes
* Support multi coordinator for v1/resourceGroupState/* endpoint

[swapsmagic]

DistributedQueryStateInfoResource to make the name consistent with the QueryInfoStateResource

[swapsmagic]

getQueryStateInfos to make the name consistent with the non distributed version.

[swapsmagic]

user info needs to be passed to the coordinator and only return queries that matches the given user.

[swapsmagic]

make the error message generic as we don't know if it's one or more Error getting query info from coordinators.

[swapsmagic]

instead of going to all coordinators we can rely on the nodeQueryStates to filter coordinators which have queries running/queued.

[swapsmagic]

do we have any use case where coordinator endpoint is called to pull entire cluster information? If not there we don't need to have this branching and just return local queries.

[swapsmagic]

ditto, if not needed, remove branching logic and just return local query info.

[swapsmagic]

Few minor comments, rest looks good.

[swapsmagic]

rename includeFromLocal to includeLocalInfoOnly

[abhiseksaikia]

done!

[swapsmagic]

i don't see us supporting xForwardedProto anywhere else. Do we know if it's been used or not?

[abhiseksaikia]

I think its not currently used in the resource group flow

[tdcmeehan]

Check the not null message here.

[tdcmeehan]

I would write this more along these lines:

            ListenableFuture<List<ResourceGroupInfo>> resourceGroupInfosFuture = Futures.allAsList(queryStateInfoFutureList);
            addSuccessCallback(resourceGroupInfosFuture, resourceGroupInfos -> {
                ResourceGroupInfo aggregatedResourceGroupInfo = buildResourceGroupInfo(resourceGroupInfos);
                if (aggregatedResourceGroupInfo == null) {
                    asyncResponse.resume(Response.status(NOT_FOUND).build());
                    return;
                }
                asyncResponse.resume(Response.ok(aggregatedResourceGroupInfo).build());
            });
            addExceptionCallback(resourceGroupInfosFuture, e -> {
                log.error(e, "Error in getting resource group info from one of the coordinators");
                asyncResponse.resume(Response.serverError().entity(e.getMessage()).build());
            });

Where buildResourceGroupInfo is a new method that takes in a list of ResourceGroupInfo and aggregates it in one shot. We can also remove AggregatedResourceGroupInfoBuilder this way.

[abhiseksaikia]

One problem I see with this is, it does not handle the case where one coordinator throws 404 and others return successful response. In this case, we still want to aggregate the successful response and return.

I think with moving the aggregation to inline method will complicate aggregation for subgroups, having a builder will avoids this complication as well as need of creating more objects.

[tdcmeehan]

See other comment: I think this class can be inlined into a method.

[tdcmeehan]

Looks like this needs a rebase?