xds: ignore keep_empty_value and discard header keys on empty mutations

In GrpcService/ext_proc header mutations, we do not support the
`keep_empty_value` field (grpc/proposal@f6d42d6).
If any header mutation results in a header containing an empty value (either string or binary), the
entire header key is discarded/removed from the metadata.

This chang modifies HeaderMutator to apply mutations and discard keys containing empty values.

Author: kannanjgithub

xds/src/main/java/io/grpc/xds/internal/headermutations/HeaderMutator.java

@@ -67,24 +67,23 @@ private void applyHeaderUpdates(final Iterable<HeaderValueOption> headerOptions,
   private void updateHeader(final HeaderValueOption option, Metadata mutableHeaders) {
     HeaderValue header = option.header();
     HeaderAppendAction action = option.appendAction();
-    boolean keepEmptyValue = option.keepEmptyValue();
 
     if (header.key().endsWith(Metadata.BINARY_HEADER_SUFFIX)) {
       if (header.rawValue().isPresent()) {
-        byte[] value = header.rawValue().get().toByteArray();
-        if (value.length > 0 || keepEmptyValue) {
-          updateHeader(action, Metadata.Key.of(header.key(), Metadata.BINARY_BYTE_MARSHALLER),
-                  value, mutableHeaders);
+        Metadata.Key<byte[]> key = Metadata.Key.of(header.key(), Metadata.BINARY_BYTE_MARSHALLER);
+        updateHeader(action, key, header.rawValue().get().toByteArray(), mutableHeaders);
+        if (containsEmpty(key, mutableHeaders)) {
+          mutableHeaders.discardAll(key);
         }
       } else {
         logger.fine("Missing binary rawValue for header: " + header.key());
       }
     } else {
       if (header.value().isPresent()) {
-        String value = header.value().get();
-        if (!value.isEmpty() || keepEmptyValue) {
-          updateHeader(action, Metadata.Key.of(header.key(), Metadata.ASCII_STRING_MARSHALLER),
-                  value, mutableHeaders);
+        Metadata.Key<String> key = Metadata.Key.of(header.key(), Metadata.ASCII_STRING_MARSHALLER);
+        updateHeader(action, key, header.value().get(), mutableHeaders);
+        if (containsEmpty(key, mutableHeaders)) {
+          mutableHeaders.discardAll(key);
         }
       } else {
         logger.fine("Missing value for header: " + header.key());
@@ -119,5 +118,24 @@ private <T> void updateHeader(final HeaderAppendAction action, final Metadata.Ke
         logger.fine("Unknown HeaderAppendAction: " + action);
     }
   }
+
+  private <T> boolean containsEmpty(Metadata.Key<T> key, Metadata headers) {
+    Iterable<T> values = headers.getAll(key);
+    if (values == null) {
+      return false;
+    }
+    for (T val : values) {
+      if (val instanceof String) {
+        if (((String) val).isEmpty()) {
+          return true;
+        }
+      } else if (val instanceof byte[]) {
+        if (((byte[]) val).length == 0) {
+          return true;
+        }
+      }
+    }
+    return false;
+  }
 }
 

xds/src/test/java/io/grpc/xds/internal/headermutations/HeaderMutatorTest.java

@@ -247,30 +247,27 @@ public void applyMutations_keepEmptyValue() {
     headerMutator.applyMutations(mutations, headers);
 
     assertThat(headers.containsKey(NEW_ADD_KEY)).isFalse();
-    assertThat(headers.getAll(APPEND_KEY)).containsExactly("existing-value");
-    assertThat(headers.get(OVERWRITE_KEY)).isEqualTo("existing-value");
+    assertThat(headers.containsKey(APPEND_KEY)).isFalse();
+    assertThat(headers.containsKey(OVERWRITE_KEY)).isFalse();
     assertThat(headers.containsKey(ADD_KEY)).isFalse();
-    assertThat(headers.get(OVERWRITE_IF_EXISTS_KEY)).isEqualTo("existing-value");
+    assertThat(headers.containsKey(OVERWRITE_IF_EXISTS_KEY)).isFalse();
 
     Metadata.Key<String> keepEmptyKey =
         Metadata.Key.of("keep-empty-key", Metadata.ASCII_STRING_MARSHALLER);
     Metadata.Key<String> keepEmptyOverwriteKey =
         Metadata.Key.of("keep-empty-overwrite-key", Metadata.ASCII_STRING_MARSHALLER);
 
-    assertThat(headers.containsKey(keepEmptyKey)).isTrue();
-    assertThat(headers.get(keepEmptyKey)).isEqualTo("");
-    assertThat(headers.containsKey(keepEmptyOverwriteKey)).isTrue();
-    assertThat(headers.get(keepEmptyOverwriteKey)).isEqualTo("");
+    assertThat(headers.containsKey(keepEmptyKey)).isFalse();
+    assertThat(headers.containsKey(keepEmptyOverwriteKey)).isFalse();
 
     Metadata.Key<byte[]> keepEmptyBinKey =
         Metadata.Key.of("keep-empty-bin-key-bin", Metadata.BINARY_BYTE_MARSHALLER);
     Metadata.Key<byte[]> ignoreEmptyBinKey =
         Metadata.Key.of("ignore-empty-bin-key-bin", Metadata.BINARY_BYTE_MARSHALLER);
 
-    assertThat(headers.containsKey(keepEmptyBinKey)).isTrue();
-    assertThat(headers.get(keepEmptyBinKey)).isEqualTo(new byte[0]);
+    assertThat(headers.containsKey(keepEmptyBinKey)).isFalse();
     assertThat(headers.containsKey(ignoreEmptyBinKey)).isFalse();
-    assertThat(headers.get(overwriteEmptyBinKey)).isEqualTo(originalBinValue);
+    assertThat(headers.containsKey(overwriteEmptyBinKey)).isFalse();
   }
 
   @Test