Address comments

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

@@ -24,6 +24,7 @@
 import com.google.common.base.Splitter;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.net.InetAddresses;
+import com.google.protobuf.Any;
 import com.google.protobuf.Struct;
 import io.grpc.Attributes;
 import io.grpc.EquivalentAddressGroup;
@@ -488,21 +489,12 @@ private SocketAddress rewriteAddress(SocketAddress addr, ImmutableMap<String,
         if (!(addr instanceof InetSocketAddress)) {
           return addr;
         }
-
-        SocketAddress proxyAddress;
-        try {
-          proxyAddress = (SocketAddress) endpointMetadata.get("envoy.http11_proxy_transport_socket.proxy_address");
-          if (proxyAddress == null) {
-            proxyAddress = (SocketAddress) localityMetadata.get("envoy.http11_proxy_transport_socket.proxy_address");
-          }
-        } catch (ClassCastException e) {
-          throw new AssertionError("Proxy address metadata is not an InetSocketAddress", e);
-        }
+        Object proxyAddress = endpointMetadata.get("envoy.http11_proxy_transport_socket.proxy_address");
 
         if (proxyAddress != null) {
           return HttpConnectProxiedSocketAddress.newBuilder()
               .setTargetAddress((InetSocketAddress) addr)
-              .setProxyAddress(proxyAddress)
+              .setProxyAddress((SocketAddress) proxyAddress)
               .build();
         }
         return addr;

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

@@ -38,6 +38,7 @@
 import io.grpc.auth.MoreCallCredentials;
 import io.grpc.xds.Filter.ClientInterceptorBuilder;
 import io.grpc.xds.MetadataRegistry.MetadataValueParser;
+import io.grpc.xds.client.XdsResourceType.ResourceInvalidException;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.concurrent.ScheduledExecutorService;
@@ -230,11 +231,16 @@ public String getTypeUrl() {
     }
 
     @Override
-    public String parse(Any any) throws InvalidProtocolBufferException {
-      Audience audience = any.unpack(Audience.class);
+    public String parse(Any any) throws ResourceInvalidException {
+      Audience audience;
+      try {
+        audience = any.unpack(Audience.class);
+      } catch (InvalidProtocolBufferException ex) {
+        throw new ResourceInvalidException("Invalid Resource in address proto", ex);
+      }
       String url = audience.getUrl();
       if (url.isEmpty()) {
-        throw new InvalidProtocolBufferException(
+        throw new ResourceInvalidException(
             "Audience URL is empty. Metadata value must contain a valid URL.");
       }
       return url;

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

@@ -24,6 +24,7 @@
 import io.envoyproxy.envoy.config.core.v3.Metadata;
 import io.grpc.xds.GcpAuthenticationFilter.AudienceMetadataParser;
 import io.grpc.xds.XdsEndpointResource.AddressMetadataParser;
+import io.grpc.xds.client.XdsResourceType.ResourceInvalidException;
 import io.grpc.xds.internal.ProtobufJsonConverter;
 import java.util.HashMap;
 import java.util.Map;
@@ -74,7 +75,7 @@ void removeParser(MetadataValueParser parser) {
    * @throws InvalidProtocolBufferException if parsing {@code typed_filter_metadata} fails.
    */
   public static ImmutableMap<String, Object> parseMetadata(Metadata metadata)
-      throws InvalidProtocolBufferException {
+      throws ResourceInvalidException, InvalidProtocolBufferException {
     ImmutableMap.Builder<String, Object> parsedMetadata = ImmutableMap.builder();
 
     MetadataRegistry registry = MetadataRegistry.getInstance();
@@ -115,6 +116,6 @@ interface MetadataValueParser {
      * @return the parsed metadata value.
      * @throws InvalidProtocolBufferException if the parsing fails.
      */
-    Object parse(Any any) throws InvalidProtocolBufferException;
+    Object parse(Any any) throws ResourceInvalidException;
   }
 }

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

@@ -295,18 +295,21 @@ public String toString() {
     }
   }
 
-  static class AddressMetadataParser implements MetadataValueParser {
+  public static class AddressMetadataParser implements MetadataValueParser {
 
     @Override
     public String getTypeUrl() {
       return "type.googleapis.com/envoy.config.core.v3.Address";
     }
 
     @Override
-    public java.net.SocketAddress parse(Any any) throws InvalidProtocolBufferException {
-      Address addressProto = any.unpack(Address.class);
-      SocketAddress socketAddress = addressProto.getSocketAddress();
-
+    public java.net.SocketAddress parse(Any any) throws ResourceInvalidException {
+      SocketAddress socketAddress;
+      try {
+        socketAddress = any.unpack(Address.class).getSocketAddress();
+      } catch (InvalidProtocolBufferException ex) {
+        throw new ResourceInvalidException("Invalid Resource in address proto", ex);
+      }
       validateAddress(socketAddress);
 
       String ip = socketAddress.getAddress();
@@ -319,18 +322,18 @@ public java.net.SocketAddress parse(Any any) throws InvalidProtocolBufferExcepti
       }
     }
 
-    private void validateAddress(SocketAddress socketAddress) throws InvalidProtocolBufferException {
+    private void validateAddress(SocketAddress socketAddress) throws ResourceInvalidException {
       if (socketAddress.getAddress().isEmpty()) {
         throw createException("Address field is empty or invalid.");
       }
       long port = Integer.toUnsignedLong(socketAddress.getPortValue());
-      if (port == 0 || port > 65535) {
+      if (port > 65535) {
         throw createException(String.format("Port value %d out of range 1-65535.", port));
       }
     }
 
-    private InvalidProtocolBufferException createException(String message) {
-      return new InvalidProtocolBufferException("Failed to parse envoy.config.core.v3.Address: " + message);
+    private ResourceInvalidException createException(String message) {
+      return new ResourceInvalidException("Failed to parse envoy.config.core.v3.Address: " + message);
     }
   }
 }

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

@@ -1198,7 +1198,7 @@ public void parseLocalityLbEndpoints_withDualStackEndpoints() {
   }
 
   @Test
-  public void parseLocalityLbEndpoints_invalidPriority() {
+  public void parseLocalityLbEndpoints_invalidPriority() throws ResourceInvalidException {
     io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints proto =
         io.envoyproxy.envoy.config.endpoint.v3.LocalityLbEndpoints.newBuilder()
             .setLocality(Locality.newBuilder()