feat: Improve YamlUtils.dumpAsMap error handling

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/utils/YamlUtils.java

@@ -21,6 +21,7 @@
 
 import java.io.InputStream;
 import java.math.BigDecimal;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -36,6 +37,7 @@
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.io.BaseEncoding;
 import org.checkerframework.checker.nullness.qual.Nullable;
 import org.yaml.snakeyaml.Yaml;
+import org.yaml.snakeyaml.error.YAMLException;
 
 public class YamlUtils {
   private static final Map<Schema.TypeName, Function<String, @Nullable Object>> YAML_VALUE_PARSERS =
@@ -181,7 +183,36 @@ public static String yamlStringFromMap(@Nullable Map<String, Object> map) {
     if (map == null || map.isEmpty()) {
       return "";
     }
-    return new Yaml().dumpAsMap(map);
+    try {
+      return new Yaml().dumpAsMap(map);
+    } catch (YAMLException e) {
+      List<String> problematicKeys = findNonSerializableKeys(map);
+      throw new IllegalArgumentException(
+          String.format(
+              "Failed to convert configuration map to YAML. "
+                  + "The following keys contain values that cannot be serialized: %s. "
+                  + "Please ensure all configuration values are simple types (String, Number, Boolean) "
+                  + "or properly structured Maps and Lists. Original error: %s",
+              problematicKeys, e.getMessage()),
+          e);
+    }
+  }
+
+  private static List<String> findNonSerializableKeys(Map<String, Object> map) {
+    List<String> problematicKeys = new ArrayList<>();
+    Yaml yaml = new Yaml();
+    for (Map.Entry<String, Object> entry : map.entrySet()) {
+      try {
+        yaml.dump(entry.getValue());
+      } catch (YAMLException e) {
+        problematicKeys.add(
+            String.format(
+                "%s (type: %s)",
+                entry.getKey(),
+                entry.getValue() != null ? entry.getValue().getClass().getName() : "null"));
+      }
+    }
+    return problematicKeys;
   }
 
   public static Map<String, Object> yamlStringToMap(@Nullable String yaml) {

sdks/java/core/src/test/java/org/apache/beam/sdk/util/YamlUtilsTest.java

@@ -258,4 +258,34 @@ public void testSnakeCaseMapToCamelCaseRow() {
 
     assertEquals(expectedRow, convertedRow);
   }
+
+  @Test
+  public void testYamlStringFromMapWithNonSerializableObject() {
+    // Create a map with ImmutableMap.Builder which cannot be serialized
+    Map<String, Object> configWithBuilder = new java.util.HashMap<>();
+    configWithBuilder.put("good_key", "valid_value");
+    configWithBuilder.put(
+        "bad_key",
+        org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap
+            .builder()); // Not yet built!
+
+    // We expect an IllegalArgumentException with a helpful message
+    thrown.expect(IllegalArgumentException.class);
+    thrown.expectMessage("Failed to convert configuration map to YAML");
+    thrown.expectMessage("bad_key");
+    thrown.expectMessage("ImmutableMap$Builder");
+
+    YamlUtils.yamlStringFromMap(configWithBuilder);
+  }
+
+  @Test
+  public void testYamlStringFromMapWithValidMap() {
+    Map<String, Object> config =
+        org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap.of(
+            "string_key", "value", "int_key", 123, "boolean_key", true);
+
+    String yaml = YamlUtils.yamlStringFromMap(config);
+    org.junit.Assert.assertNotNull(yaml);
+    org.junit.Assert.assertTrue(yaml.contains("string_key"));
+  }
 }