Review points

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

@@ -22,6 +22,7 @@
 import io.grpc.xds.GcpAuthenticationFilter.AudienceMetadataParser;
 import java.util.HashMap;
 import java.util.Map;
+import javax.annotation.Nullable;
 
 /**
  * Registry for parsing cluster metadata values.
@@ -42,6 +43,7 @@ static ClusterMetadataRegistry getInstance() {
     return INSTANCE;
   }
 
+  @Nullable
   ClusterMetadataValueParser findParser(String typeUrl) {
     return supportedParsers.get(typeUrl);
   }
@@ -51,6 +53,10 @@ void registerParser(ClusterMetadataValueParser parser) {
     supportedParsers.put(parser.getTypeUrl(), parser);
   }
 
+  void removeParser(ClusterMetadataValueParser parser) {
+    supportedParsers.remove(parser.getTypeUrl());
+  }
+
   interface ClusterMetadataValueParser {
 
     String getTypeUrl();

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

@@ -222,9 +222,6 @@ V getOrInsert(K key, Function<K, V> create) {
     }
   }
 
-  /**
-   * Parser for Audience metadata type.
-   */
   static class AudienceMetadataParser implements ClusterMetadataValueParser {
 
     @Override
@@ -234,19 +231,13 @@ public String getTypeUrl() {
 
     @Override
     public String parse(Any any) throws InvalidProtocolBufferException {
-      if (any.is(Audience.class)) {
-        Audience audience = any.unpack(Audience.class);
-        String url = audience.getUrl();
-        if (url.isEmpty()) {
-          throw new InvalidProtocolBufferException(
-              "Audience URL is empty. Metadata value must contain a valid URL.");
-        }
-        return url;
-      } else {
+      Audience audience = any.unpack(Audience.class);
+      String url = audience.getUrl();
+      if (url.isEmpty()) {
         throw new InvalidProtocolBufferException(
-            String.format("Unexpected message type: %s. Expected: %s",
-                any.getTypeUrl(), Audience.getDescriptor().getFullName()));
+            "Audience URL is empty. Metadata value must contain a valid URL.");
       }
+      return url;
     }
   }
 }

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

@@ -187,6 +187,18 @@ static CdsUpdate processCluster(Cluster cluster,
     return updateBuilder.build();
   }
 
+  /**
+   * Parses cluster metadata into a structured map.
+   *
+   * <p>Values in {@code typed_filter_metadata} take precedence over
+   * {@code filter_metadata} when keys overlap, following Envoy API behavior. See
+   * <a href="https://github.com/envoyproxy/envoy/blob/main/api/envoy/config/core/v3/base.proto#L217-L259">
+   *   Envoy metadata documentation </a> for details.
+   *
+   * @param metadata the {@link Metadata} containing the fields to parse.
+   * @return an immutable map of parsed metadata.
+   * @throws InvalidProtocolBufferException if parsing {@code typed_filter_metadata} fails.
+   */
   private static ImmutableMap<String, Object> parseClusterMetadata(Metadata metadata)
       throws InvalidProtocolBufferException {
     ImmutableMap.Builder<String, Object> parsedMetadata = ImmutableMap.builder();

xds/src/main/java/io/grpc/xds/internal/ProtobufJsonConverter.java

@@ -22,6 +22,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.stream.Collectors;
+import javax.annotation.Nullable;
 
 /**
  * Converter for Protobuf {@link Struct} to JSON-like {@link Map}.
@@ -38,7 +39,8 @@ public static Map<String, Object> convertToJson(Struct struct) {
     return result;
   }
 
-  public static Object convertValue(Value value) {
+  @Nullable
+  private static Object convertValue(Value value) {
     switch (value.getKindCase()) {
       case STRUCT_VALUE:
         return convertToJson(value.getStructValue());

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

@@ -124,6 +124,7 @@
 import io.grpc.lookup.v1.NameMatcher;
 import io.grpc.lookup.v1.RouteLookupClusterSpecifier;
 import io.grpc.lookup.v1.RouteLookupConfig;
+import io.grpc.xds.ClusterMetadataRegistry.ClusterMetadataValueParser;
 import io.grpc.xds.ClusterSpecifierPlugin.NamedPluginConfig;
 import io.grpc.xds.ClusterSpecifierPlugin.PluginConfig;
 import io.grpc.xds.Endpoints.LbEndpoint;
@@ -2347,18 +2348,22 @@ public void parseCluster_validateEdsSourceConfig() throws ResourceInvalidExcepti
   @Test
   public void processCluster_parsesMetadata()
       throws ResourceInvalidException, InvalidProtocolBufferException {
-    ClusterMetadataRegistry testRegistry = ClusterMetadataRegistry.getInstance();
-    testRegistry.registerParser(new ClusterMetadataRegistry.ClusterMetadataValueParser() {
-      @Override
-      public String getTypeUrl() {
-        return "type.googleapis.com/test.Type";
-      }
-
-      @Override
-      public Object parse(Any value) {
-        return "testValue";
-      }
-    });
+    ClusterMetadataRegistry clusterMetadataRegistry = ClusterMetadataRegistry.getInstance();
+
+    ClusterMetadataValueParser testParser =
+        new ClusterMetadataRegistry.ClusterMetadataValueParser() {
+          @Override
+          public String getTypeUrl() {
+            return "type.googleapis.com/test.Type";
+          }
+
+          @Override
+          public Object parse(Any value) {
+            assertThat(value.getValue().toStringUtf8()).isEqualTo("test");
+            return value.getValue().toStringUtf8() + "_processed";
+          }
+        };
+    clusterMetadataRegistry.registerParser(testParser);
 
     Any typedFilterMetadata = Any.newBuilder()
         .setTypeUrl("type.googleapis.com/test.Type")
@@ -2393,11 +2398,12 @@ public Object parse(Any value) {
         LoadBalancerRegistry.getDefaultRegistry());
 
     ImmutableMap<String, Object> expectedParsedMetadata = ImmutableMap.of(
-        "TYPED_FILTER_METADATA", "testValue",
+        "TYPED_FILTER_METADATA", "test_processed",
         "FILTER_METADATA", ImmutableMap.of(
             "key1", "value1",
             "key2", 42.0));
     assertThat(update.parsedMetadata()).isEqualTo(expectedParsedMetadata);
+    clusterMetadataRegistry.removeParser(testParser);
   }
 
   @Test
@@ -2440,8 +2446,6 @@ public void processCluster_parsesAudienceMetadata()
     CdsUpdate update = XdsClusterResource.processCluster(
         cluster, null, LRS_SERVER_INFO,
         LoadBalancerRegistry.getDefaultRegistry());
-    System.out.println("XXXXX");
-    System.out.println(update.parsedMetadata());
 
     ImmutableMap<String, Object> expectedParsedMetadata = ImmutableMap.of(
         "AUDIENCE_METADATA", "https://example.com",
@@ -2451,6 +2455,63 @@ public void processCluster_parsesAudienceMetadata()
     assertThat(update.parsedMetadata()).isEqualTo(expectedParsedMetadata);
   }
 
+  @Test
+  public void processCluster_metadataKeyCollision_resolvesToTypedMetadata()
+      throws ResourceInvalidException, InvalidProtocolBufferException {
+    ClusterMetadataRegistry clusterMetadataRegistry = ClusterMetadataRegistry.getInstance();
+
+    ClusterMetadataValueParser testParser =
+        new ClusterMetadataRegistry.ClusterMetadataValueParser() {
+          @Override
+          public String getTypeUrl() {
+            return "type.googleapis.com/test.Type";
+          }
+
+          @Override
+          public Object parse(Any value) {
+            return "typedMetadataValue";
+          }
+        };
+    clusterMetadataRegistry.registerParser(testParser);
+
+    Any typedFilterMetadata = Any.newBuilder()
+        .setTypeUrl("type.googleapis.com/test.Type")
+        .setValue(ByteString.copyFromUtf8("test"))
+        .build();
+
+    Struct filterMetadata = Struct.newBuilder()
+        .putFields("key1", Value.newBuilder().setStringValue("filterMetadataValue").build())
+        .build();
+
+    Metadata metadata = Metadata.newBuilder()
+        .putTypedFilterMetadata("key1", typedFilterMetadata)
+        .putFilterMetadata("key1", filterMetadata)
+        .build();
+
+    Cluster cluster = Cluster.newBuilder()
+        .setName("cluster-foo.googleapis.com")
+        .setType(DiscoveryType.EDS)
+        .setEdsClusterConfig(
+            EdsClusterConfig.newBuilder()
+                .setEdsConfig(
+                    ConfigSource.newBuilder()
+                        .setAds(AggregatedConfigSource.getDefaultInstance()))
+                .setServiceName("service-foo.googleapis.com"))
+        .setLbPolicy(LbPolicy.ROUND_ROBIN)
+        .setMetadata(metadata)
+        .build();
+
+    CdsUpdate update = XdsClusterResource.processCluster(
+        cluster, null, LRS_SERVER_INFO,
+        LoadBalancerRegistry.getDefaultRegistry());
+
+    ImmutableMap<String, Object> expectedParsedMetadata = ImmutableMap.of(
+        "key1", "typedMetadataValue");
+    assertThat(update.parsedMetadata()).isEqualTo(expectedParsedMetadata);
+    clusterMetadataRegistry.removeParser(testParser);
+  }
+
+
   @Test
   public void parseServerSideListener_invalidTrafficDirection() throws ResourceInvalidException {
     Listener listener =