lost watch when client send two request without nonce

Start at latestDiscoveryResponse, resourceNonce, and computeWatch in the handling shown in the issue; reproduce the restart case with a C++ client sending two requests without a nonce for one resource type. Trace why the second request is discarded, then verify that both requests' watched resources are retained with a regression test.

log

[2022-04-25 19:13:25.993] [DEBUG] [grpc-default-executor-0] [i.e.controlplane.server.OspDiscoveryServer] >>> [0] open stream from 
[2022-04-25 19:13:26.009] [DEBUG] [grpc-default-executor-0] [i.e.controlplane.server.OspDiscoveryServer] >>> [0] request type.googleapis.com/envoy.api.v2.ClusterLoadAssignment[r1] with nonce  from version 3
[2022-04-25 19:13:26.684] [DEBUG] [grpc-default-executor-0] [i.e.controlplane.server.OspDiscoveryServer] >>> [0] response type.googleapis.com/envoy.api.v2.ClusterLoadAssignment with nonce 0 version 0
[2022-04-25 19:13:26.696] [DEBUG] [grpc-default-executor-0] [i.e.controlplane.server.OspDiscoveryServer] >>> [0] request type.googleapis.com/envoy.api.v2.ClusterLoadAssignment[r1, r2, r3, r4] with nonce  from version 3

When I restart xds server, C++ client will send two request without nonce for one resourceType, xds server response first request, then handle the second request.

In the code, xds server will discard the second request, watch resources in second request will lost.

    LatestDiscoveryResponse latestDiscoveryResponse = latestResponse(requestTypeUrl);
    String resourceNonce = latestDiscoveryResponse == null ? null : latestDiscoveryResponse.nonce();

    if (isNullOrEmpty(resourceNonce) || resourceNonce.equals(nonce)) {
      if (!request.hasErrorDetail() && latestDiscoveryResponse != null) {
        setAckedResources(requestTypeUrl, latestDiscoveryResponse.resourceNames());
      }

      computeWatch(requestTypeUrl, () -> discoveryServer.configWatcher.createWatch(
          ads(),
          request,
          ackedResources(requestTypeUrl),
          r -> executor.execute(() -> send(r, requestTypeUrl)),
          hasClusterChanged
      ));
    }