xds: Reject duplicate endpoint addresses in EDS (#13127)

Fixes #13119.

Reject duplicate endpoint socket addresses while parsing EDS resources.
  Previously, these resources were ACKed and could trigger a panic in
  outlier detection.

Track addresses across the original locality entries, including entries
  that may later be overwritten in the locality-keyed output map. Reject
duplicates via ResourceInvalidException so the existing NACK path
retains
  the last accepted resource and version.

  Validation includes additional addresses when dual-stack is enabled.
  The same IP with different ports remains valid, and zero-weight
  localities remain ignored.

  Regression tests cover duplicates within and across localities and
  priorities, overwritten locality entries, equivalent IPv6 addresses,
  dual-stack addresses, cold start, repeated NACKs, and recovery.

  ### Testing

  - Original parser: 14 of 18 new parameterized regression cases fail.
  - Patched parser: all 272 cases in GrpcXdsClientImplV3Test pass using
    a standalone JUnit harness with grpc-java 1.85.0 dependencies.
  - Checkstyle 10.26.1 passes for both modified files.

  ### Gradle build success

```
[16:17:13] /Users/valter/projects/grpc-java-pr git:(fix-13119-duplicate-eds-endpoints) > ./gradlew -PskipCodegen=true -PskipAndroid=true \
    :grpc-xds:test --tests io.grpc.xds.GrpcXdsClientImplV3Test \
    :grpc-xds:checkstyleMain :grpc-xds:checkstyleTest

Welcome to Gradle 8.14.5!

Here are the highlights of this release:
 - Java 24 support
 - GraalVM Native Image toolchain selection
 - Enhancements to test reporting
 - Build Authoring improvements

For more details see https://docs.gradle.org/8.14.5/release-notes.html

Starting a Gradle Daemon (subsequent builds will be faster)
*** Skipping the build of codegen and compilation of proto files because skipCodegen=true
  * Skipping the build of Android projects because skipAndroid=true

> Task :grpc-services:compileJava
Note: Some input files use or override a deprecated API.
Note: Recompile with -Xlint:deprecation for details.

[Incubating] Problems report is available at: file:///Users/valter/projects/grpc-java-pr/build/reports/problems/problems-report.html

Deprecated Gradle features were used in this build, making it incompatible with Gradle 9.0.

You can use '--warning-mode all' to show the individual deprecation warnings and determine if they come from your own scripts or plugins.

For more on this, please refer to https://docs.gradle.org/8.14.5/userguide/command_line_interface.html#sec:command_line_warnings in the Gradle documentation.

BUILD SUCCESSFUL in 3m 9s
83 actionable tasks: 83 executed
```

Author: valeter

xds/src/main/java/io/grpc/xds/XdsEndpointResource.java

@@ -116,6 +116,7 @@ private static EdsUpdate processClusterLoadAssignment(ClusterLoadAssignment assi
       throws ResourceInvalidException {
     Map<Integer, Set<Locality>> priorities = new HashMap<>();
     Map<Locality, LocalityLbEndpoints> localityLbEndpointsMap = new LinkedHashMap<>();
+    Set<java.net.SocketAddress> endpointAddresses = new HashSet<>();
     List<Endpoints.DropOverload> dropOverloads = new ArrayList<>();
     int maxPriority = -1;
     for (io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints localityLbEndpointsProto
@@ -144,6 +145,14 @@ private static EdsUpdate processClusterLoadAssignment(ClusterLoadAssignment assi
         throw new ResourceInvalidException("ClusterLoadAssignment has duplicate locality:"
             + locality + " for priority:" + priority);
       }
+      for (Endpoints.LbEndpoint endpoint : localityLbEndpoints.endpoints()) {
+        for (java.net.SocketAddress address : endpoint.eag().getAddresses()) {
+          if (!endpointAddresses.add(address)) {
+            throw new ResourceInvalidException(
+                "ClusterLoadAssignment has duplicate endpoint address: " + address);
+          }
+        }
+      }
     }
     if (priorities.size() != maxPriority + 1) {
       throw new ResourceInvalidException("ClusterLoadAssignment has sparse priorities");

xds/src/test/java/io/grpc/xds/ClusterResolverLoadBalancerTest.java

@@ -682,11 +682,15 @@ public void edsUpdatePriorityName_mergeTwoPriorities() {
   }
 
   private LocalityLbEndpoints createEndpoints(Locality locality, int priority) {
+    // Localities must have distinct endpoint addresses even when their priorities are merged.
+    String address = ImmutableMap.of(
+        LOCALITY1, "127.0.0.1", LOCALITY2, "127.0.0.2", LOCALITY3, "127.0.0.3")
+        .get(locality);
     return LocalityLbEndpoints.newBuilder()
         .setLoadBalancingWeight(UInt32Value.of(70))
         .setLocality(locality)
         .setPriority(priority)
-        .addLbEndpoints(newSocketLbEndpoint("127.0." + priority + ".1", 8080))
+        .addLbEndpoints(newSocketLbEndpoint(address, 8080))
         .build();
   }
 

xds/src/test/java/io/grpc/xds/GrpcXdsClientImplTestBase.java

@@ -19,6 +19,7 @@
 import static com.google.common.truth.Truth.assertThat;
 import static com.google.common.truth.Truth.assertWithMessage;
 import static io.grpc.StatusMatcher.statusHasCode;
+import static io.grpc.xds.XdsEndpointResource.GRPC_EXPERIMENTAL_XDS_DUALSTACK_ENDPOINTS;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.argThat;
 import static org.mockito.ArgumentMatchers.eq;
@@ -47,6 +48,9 @@
 import com.google.protobuf.UInt32Value;
 import com.google.protobuf.util.Durations;
 import io.envoyproxy.envoy.config.cluster.v3.OutlierDetection;
+import io.envoyproxy.envoy.config.core.v3.Address;
+import io.envoyproxy.envoy.config.core.v3.SocketAddress;
+import io.envoyproxy.envoy.config.endpoint.v3.Endpoint;
 import io.envoyproxy.envoy.config.route.v3.FilterConfig;
 import io.envoyproxy.envoy.config.route.v3.WeightedCluster;
 import io.envoyproxy.envoy.extensions.filters.http.router.v3.Router;
@@ -4068,6 +4072,215 @@ public void edsResourceUpdated() {
     verifySubscribedResourcesMetadataSizes(0, 0, 0, 1);
   }
 
+  @Test
+  public void edsDuplicateEndpointAcrossLocalities_retainsLastGoodResource() {
+    DiscoveryRpcCall call = startResourceWatcher(XdsEndpointResource.getInstance(), EDS_RESOURCE,
+        edsResourceWatcher);
+    call.sendResponse(EDS, testClusterLoadAssignment, VERSION_1, "0000");
+    call.verifyRequest(EDS, EDS_RESOURCE, VERSION_1, "0000", NODE);
+    verify(edsResourceWatcher).onResourceChanged(edsUpdateCaptor.capture());
+    EdsUpdate lastGoodUpdate = edsUpdateCaptor.getValue().getValue();
+
+    Any invalidAssignment = Any.pack(mf.buildClusterLoadAssignment(EDS_RESOURCE,
+        ImmutableList.of(
+            mf.buildLocalityLbEndpoints("region1", "zone1", "subzone1",
+                mf.buildLbEndpoint("172.44.2.2", 8000, "healthy", 1, "host1"), 1, 0),
+            mf.buildLocalityLbEndpoints("region2", "zone2", "subzone2",
+                mf.buildLbEndpoint("172.44.2.2", 8000, "unhealthy", 2, "host2"), 1, 0)),
+        ImmutableList.of()));
+    List<String> errors = ImmutableList.of("EDS response ClusterLoadAssignment '" + EDS_RESOURCE
+        + "' validation error: ClusterLoadAssignment has duplicate endpoint address: "
+        + "/172.44.2.2:8000");
+    call.sendResponse(EDS, invalidAssignment, VERSION_2, "0001");
+    call.verifyRequestNack(EDS, EDS_RESOURCE, VERSION_1, "0001", NODE, errors);
+    verifyResourceMetadataNacked(EDS, EDS_RESOURCE, testClusterLoadAssignment, VERSION_1,
+        TIME_INCREMENT, VERSION_2, TIME_INCREMENT * 2, errors, true);
+    ArgumentCaptor<Status> statusCaptor = ArgumentCaptor.forClass(Status.class);
+    verify(edsResourceWatcher).onAmbientError(statusCaptor.capture());
+    assertThat(statusCaptor.getValue().getCode()).isEqualTo(Code.UNAVAILABLE);
+    assertThat(statusCaptor.getValue().getDescription()).contains(errors.get(0));
+    verifyNoMoreInteractions(edsResourceWatcher);
+
+    // A new watcher still receives the last accepted resource.
+    @SuppressWarnings("unchecked")
+    ResourceWatcher<EdsUpdate> watcher = mock(ResourceWatcher.class);
+    xdsClient.watchXdsResource(XdsEndpointResource.getInstance(), EDS_RESOURCE, watcher);
+    verify(watcher).onResourceChanged(StatusOr.fromValue(lastGoodUpdate));
+    verify(watcher).onAmbientError(statusCaptor.getValue());
+
+    call.sendResponse(EDS, invalidAssignment, VERSION_3, "0002");
+    call.verifyRequestNack(EDS, EDS_RESOURCE, VERSION_1, "0002", NODE, errors);
+    verifyResourceMetadataNacked(EDS, EDS_RESOURCE, testClusterLoadAssignment, VERSION_1,
+        TIME_INCREMENT, VERSION_3, TIME_INCREMENT * 3, errors, true);
+    verify(edsResourceWatcher, times(2)).onAmbientError(any());
+    verify(watcher, times(2)).onAmbientError(any());
+    verifyNoMoreInteractions(edsResourceWatcher, watcher);
+
+    Any validAssignment = Any.pack(mf.buildClusterLoadAssignment(EDS_RESOURCE,
+        ImmutableList.of(mf.buildLocalityLbEndpoints("region2", "zone2", "subzone2",
+            mf.buildLbEndpoint("172.44.2.2", 8001, "healthy", 1, "host2"), 1, 0)),
+        ImmutableList.of()));
+    call.sendResponse(EDS, validAssignment, "45", "0003");
+    call.verifyRequest(EDS, EDS_RESOURCE, "45", "0003", NODE);
+    verifyResourceMetadataAcked(EDS, EDS_RESOURCE, validAssignment, "45", TIME_INCREMENT * 4);
+    verify(edsResourceWatcher, times(2)).onResourceChanged(edsUpdateCaptor.capture());
+    assertThat(edsUpdateCaptor.getValue().getValue()).isNotEqualTo(lastGoodUpdate);
+    verify(watcher).onResourceChanged(edsUpdateCaptor.getValue());
+  }
+
+  @Test
+  public void edsDuplicateEndpointOnColdStart() {
+    DiscoveryRpcCall call = startResourceWatcher(XdsEndpointResource.getInstance(), EDS_RESOURCE,
+        edsResourceWatcher);
+    Any invalidAssignment = Any.pack(mf.buildClusterLoadAssignment(EDS_RESOURCE,
+        ImmutableList.of(
+            mf.buildLocalityLbEndpoints("region1", "zone1", "subzone1",
+                mf.buildLbEndpoint("172.44.2.2", 8000, "healthy", 1, "host1"), 1, 0),
+            mf.buildLocalityLbEndpoints("region2", "zone2", "subzone2",
+                mf.buildLbEndpoint("172.44.2.2", 8000, "healthy", 1, "host2"), 1, 0)),
+        ImmutableList.of()));
+    call.sendResponse(EDS, invalidAssignment, VERSION_1, "0000");
+    call.verifyRequestNack(EDS, EDS_RESOURCE, "", "0000", NODE,
+        ImmutableList.of("EDS response ClusterLoadAssignment '" + EDS_RESOURCE
+            + "' validation error: ClusterLoadAssignment has duplicate endpoint address:"));
+    verify(edsResourceWatcher).onResourceChanged(edsUpdateCaptor.capture());
+    assertThat(edsUpdateCaptor.getValue().hasValue()).isFalse();
+
+    call.sendResponse(EDS, testClusterLoadAssignment, VERSION_2, "0001");
+    call.verifyRequest(EDS, EDS_RESOURCE, VERSION_2, "0001", NODE);
+    verify(edsResourceWatcher, times(2)).onResourceChanged(edsUpdateCaptor.capture());
+    assertThat(edsUpdateCaptor.getValue().hasValue()).isTrue();
+    validateGoldenClusterLoadAssignment(edsUpdateCaptor.getValue().getValue());
+  }
+
+  @Test
+  public void edsDuplicateEndpointWithinLocality() {
+    verifyEdsDuplicateEndpoint(ImmutableList.of(
+        mf.buildLocalityLbEndpoints("region1", "zone1", "subzone1",
+            ImmutableList.of(
+                mf.buildLbEndpoint("172.44.2.2", 8000, "healthy", 1, "host1"),
+                mf.buildLbEndpoint("172.44.2.2", 8000, "healthy", 1, "host2")), 1, 0)));
+  }
+
+  @Test
+  public void edsDuplicateEndpointAcrossPriorities() {
+    verifyEdsDuplicateEndpoint(ImmutableList.of(
+        mf.buildLocalityLbEndpoints("region1", "zone1", "subzone1",
+            mf.buildLbEndpoint("172.44.2.2", 8000, "healthy", 1, "host1"), 1, 0),
+        mf.buildLocalityLbEndpoints("region2", "zone2", "subzone2",
+            mf.buildLbEndpoint("172.44.2.2", 8000, "healthy", 1, "host2"), 1, 1)));
+  }
+
+  @Test
+  public void edsDuplicateEndpointInOverwrittenLocality() {
+    // The second entry overwrites the first in the locality-keyed output map.
+    verifyEdsDuplicateEndpoint(ImmutableList.of(
+        mf.buildLocalityLbEndpoints("region1", "zone1", "subzone1",
+            mf.buildLbEndpoint("172.44.2.2", 8000, "healthy", 1, "host1"), 1, 0),
+        mf.buildLocalityLbEndpoints("region1", "zone1", "subzone1",
+            mf.buildLbEndpoint("172.44.2.3", 8000, "healthy", 1, "host2"), 1, 1),
+        mf.buildLocalityLbEndpoints("region2", "zone2", "subzone2",
+            mf.buildLbEndpoint("172.44.2.2", 8000, "healthy", 1, "host3"), 1, 0)));
+  }
+
+  @Test
+  public void edsDuplicateEndpointWithEquivalentIpv6Addresses() {
+    verifyEdsDuplicateEndpoint(ImmutableList.of(
+        mf.buildLocalityLbEndpoints("region1", "zone1", "subzone1",
+            mf.buildLbEndpoint("2001:db8::1", 8000, "healthy", 1, "host1"), 1, 0),
+        mf.buildLocalityLbEndpoints("region2", "zone2", "subzone2",
+            mf.buildLbEndpoint("2001:0db8:0:0:0:0:0:1", 8000, "healthy", 1, "host2"), 1, 0)));
+  }
+
+  @Test
+  public void edsDuplicateEndpointInAdditionalAddresses() {
+    String originalDualStackProp =
+        System.setProperty(GRPC_EXPERIMENTAL_XDS_DUALSTACK_ENDPOINTS, "false");
+    try {
+      io.envoyproxy.envoy.config.endpoint.v3.LbEndpoint endpoint =
+          (io.envoyproxy.envoy.config.endpoint.v3.LbEndpoint)
+              mf.buildLbEndpoint("172.44.2.2", 8000, "healthy", 1, "host1");
+      Message dualStackEndpoint = endpoint.toBuilder()
+          .setEndpoint(endpoint.getEndpoint().toBuilder()
+              .addAdditionalAddresses(Endpoint.AdditionalAddress.newBuilder()
+                  .setAddress(Address.newBuilder().setSocketAddress(SocketAddress.newBuilder()
+                      .setAddress("2001:db8::1").setPortValue(8000)))))
+          .build();
+      Any assignment = Any.pack(mf.buildClusterLoadAssignment(EDS_RESOURCE,
+          ImmutableList.of(
+              mf.buildLocalityLbEndpoints("region1", "zone1", "subzone1",
+                  dualStackEndpoint, 1, 0),
+              mf.buildLocalityLbEndpoints("region2", "zone2", "subzone2",
+                  mf.buildLbEndpoint("2001:db8::1", 8000, "healthy", 1, "host2"), 1, 0)),
+          ImmutableList.of()));
+      DiscoveryRpcCall call = startResourceWatcher(XdsEndpointResource.getInstance(), EDS_RESOURCE,
+          edsResourceWatcher);
+      call.sendResponse(EDS, assignment, VERSION_1, "0000");
+      call.verifyRequest(EDS, EDS_RESOURCE, VERSION_1, "0000", NODE);
+
+      System.setProperty(GRPC_EXPERIMENTAL_XDS_DUALSTACK_ENDPOINTS, "true");
+      call.sendResponse(EDS, assignment, VERSION_2, "0001");
+      List<String> errors = ImmutableList.of("EDS response ClusterLoadAssignment '" + EDS_RESOURCE
+          + "' validation error: ClusterLoadAssignment has duplicate endpoint address:");
+      call.verifyRequestNack(EDS, EDS_RESOURCE, VERSION_1, "0001", NODE, errors);
+      verifyResourceMetadataNacked(EDS, EDS_RESOURCE, assignment, VERSION_1, TIME_INCREMENT,
+          VERSION_2, TIME_INCREMENT * 2, errors, true);
+      verify(edsResourceWatcher).onResourceChanged(edsUpdateCaptor.capture());
+      assertThat(edsUpdateCaptor.getValue().hasValue()).isTrue();
+    } finally {
+      if (originalDualStackProp == null) {
+        System.clearProperty(GRPC_EXPERIMENTAL_XDS_DUALSTACK_ENDPOINTS);
+      } else {
+        System.setProperty(GRPC_EXPERIMENTAL_XDS_DUALSTACK_ENDPOINTS, originalDualStackProp);
+      }
+    }
+  }
+
+  private void verifyEdsDuplicateEndpoint(List<Message> localities) {
+    DiscoveryRpcCall call = startResourceWatcher(XdsEndpointResource.getInstance(), EDS_RESOURCE,
+        edsResourceWatcher);
+    Any assignment = Any.pack(mf.buildClusterLoadAssignment(EDS_RESOURCE, localities,
+        ImmutableList.of()));
+    call.sendResponse(EDS, assignment, VERSION_1, "0000");
+    call.verifyRequestNack(EDS, EDS_RESOURCE, "", "0000", NODE,
+        ImmutableList.of("EDS response ClusterLoadAssignment '" + EDS_RESOURCE
+            + "' validation error: ClusterLoadAssignment has duplicate endpoint address:"));
+    verify(edsResourceWatcher).onResourceChanged(edsUpdateCaptor.capture());
+    assertThat(edsUpdateCaptor.getValue().hasValue()).isFalse();
+  }
+
+  @Test
+  public void edsEndpointsWithSameIpAndDifferentPorts() {
+    DiscoveryRpcCall call = startResourceWatcher(XdsEndpointResource.getInstance(), EDS_RESOURCE,
+        edsResourceWatcher);
+    Any assignment = Any.pack(mf.buildClusterLoadAssignment(EDS_RESOURCE,
+        ImmutableList.of(
+            mf.buildLocalityLbEndpoints("region1", "zone1", "subzone1",
+                mf.buildLbEndpoint("172.44.2.2", 8000, "healthy", 1, "host1"), 1, 0),
+            mf.buildLocalityLbEndpoints("region2", "zone2", "subzone2",
+                mf.buildLbEndpoint("172.44.2.2", 8001, "healthy", 1, "host2"), 1, 0)),
+        ImmutableList.of()));
+    call.sendResponse(EDS, assignment, VERSION_1, "0000");
+    call.verifyRequest(EDS, EDS_RESOURCE, VERSION_1, "0000", NODE);
+    verify(edsResourceWatcher).onResourceChanged(edsUpdateCaptor.capture());
+    assertThat(edsUpdateCaptor.getValue().getValue().localityLbEndpointsMap).hasSize(2);
+  }
+
+  @Test
+  public void edsDuplicateEndpointInZeroWeightLocalityIgnored() {
+    DiscoveryRpcCall call = startResourceWatcher(XdsEndpointResource.getInstance(), EDS_RESOURCE,
+        edsResourceWatcher);
+    Any assignment = Any.pack(mf.buildClusterLoadAssignment(EDS_RESOURCE,
+        ImmutableList.of(lbEndpointHealthy,
+            mf.buildLocalityLbEndpoints("region2", "zone2", "subzone2",
+                mf.buildLbEndpoint("192.168.0.1", 8080, "healthy", 1, "host2"), 0, 0)),
+        ImmutableList.of()));
+    call.sendResponse(EDS, assignment, VERSION_1, "0000");
+    call.verifyRequest(EDS, EDS_RESOURCE, VERSION_1, "0000", NODE);
+    verify(edsResourceWatcher).onResourceChanged(edsUpdateCaptor.capture());
+    assertThat(edsUpdateCaptor.getValue().getValue().localityLbEndpointsMap).hasSize(1);
+  }
+
   @Test
   public void edsDuplicateLocalityInTheSamePriority() {
     DiscoveryRpcCall call = startResourceWatcher(XdsEndpointResource.getInstance(), EDS_RESOURCE,