Allow empty string value for ValueProvider types.

sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptionsFactory.java

@@ -55,7 +55,6 @@
 import java.lang.annotation.Annotation;
 import java.lang.reflect.Method;
 import java.lang.reflect.Modifier;
-import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Proxy;
 import java.lang.reflect.Type;
 import java.util.ArrayList;
@@ -1610,13 +1609,7 @@ public boolean apply(@Nonnull String input) {
               throw new IllegalArgumentException(msg, e);
             }
           }
-        } else if ((returnType.isArray() && (SIMPLE_TYPES.contains(returnType.getComponentType())
-                   || returnType.getComponentType().isEnum()))
-                   || Collection.class.isAssignableFrom(returnType)
-                   || (returnType.equals(ValueProvider.class)
-                       && MAPPER.getTypeFactory().constructType(
-                         ((ParameterizedType) method.getGenericReturnType())
-                         .getActualTypeArguments()[0]).isCollectionLikeType())) {
+        } else if (isCollectionOrArrayOfAllowedTypes(returnType, type)) {
           // Split any strings with ","
           List<String> values = FluentIterable.from(entry.getValue())
               .transformAndConcat(new Function<String, Iterable<String>>() {
@@ -1626,31 +1619,21 @@ public Iterable<String> apply(@Nonnull String input) {
                 }
           }).toList();
 
-          if (returnType.isArray() && !returnType.getComponentType().equals(String.class)
-              || Collection.class.isAssignableFrom(returnType)
-              || returnType.equals(ValueProvider.class)) {
-            for (String value : values) {
-              checkArgument(!value.isEmpty(),
-                  "Empty argument value is only allowed for String, String Array, "
-                            + "and Collections of Strings, but received: %s",
-                            method.getGenericReturnType());
-            }
+          if (values.contains("")) {
+            checkEmptyStringAllowed(returnType, type, method.getGenericReturnType().toString());
           }
           convertedOptions.put(entry.getKey(), MAPPER.convertValue(values, type));
-        } else if (SIMPLE_TYPES.contains(returnType) || returnType.isEnum()
-                   || returnType.equals(ValueProvider.class)) {
+        } else if (isSimpleType(returnType, type)) {
           String value = Iterables.getOnlyElement(entry.getValue());
-          checkArgument(returnType.equals(String.class) || !value.isEmpty(),
-               "Empty argument value is only allowed for String, String Array, "
-                        + "and Collections of Strings, but received: %s",
-                        method.getGenericReturnType());
+          if (value.isEmpty()) {
+            checkEmptyStringAllowed(returnType, type, method.getGenericReturnType().toString());
+          }
           convertedOptions.put(entry.getKey(), MAPPER.convertValue(value, type));
         } else {
           String value = Iterables.getOnlyElement(entry.getValue());
-          checkArgument(returnType.equals(String.class) || !value.isEmpty(),
-                "Empty argument value is only allowed for String, String Array, "
-                        + "and Collections of Strings, but received: %s",
-                        method.getGenericReturnType());
+          if (value.isEmpty()) {
+            checkEmptyStringAllowed(returnType, type, method.getGenericReturnType().toString());
+          }
           try {
             convertedOptions.put(entry.getKey(), MAPPER.readValue(value, type));
           } catch (IOException e) {
@@ -1669,6 +1652,77 @@ public Iterable<String> apply(@Nonnull String input) {
     return convertedOptions;
   }
 
+  /**
+   * Returns true if the given type is a SIMPLE_TYPES, enum or any of these types in a
+   * parameterized ValueProvider.
+   */
+  private static boolean isSimpleType(Class<?> type, JavaType genericType) {
+    Class<?> unwrappedType = type.equals(ValueProvider.class)
+        ? genericType.containedType(0).getRawClass() : type;
+    return SIMPLE_TYPES.contains(unwrappedType) || unwrappedType.isEnum();
+  }
+
+  /**
+   * Returns true if the given type is a any Array or Collection of SIMPLE_TYPES or enum, or
+   * any of these types in a parameterized ValueProvider.
+   */
+  private static boolean isCollectionOrArrayOfAllowedTypes(Class<?> type, JavaType genericType) {
+    Class<?> containerType = type.equals(ValueProvider.class)
+        ? genericType.containedType(0).getRawClass() : type;
+
+    // Check if it is an array of simple types or enum.
+    if (containerType.isArray() && (SIMPLE_TYPES.contains(containerType.getComponentType())
+        || containerType.getComponentType().isEnum())) {
+        return true;
+    }
+    // Check if it is Collection of simple types or enum.
+    if (Collection.class.isAssignableFrom(containerType)) {
+      JavaType innerType = type.equals(ValueProvider.class)
+          ? genericType.containedType(0).containedType(0)
+          : genericType.containedType(0);
+      // Note that raw types are allowed, hence the null check.
+      if (innerType == null || SIMPLE_TYPES.contains(innerType.getRawClass())
+          || innerType.getRawClass().isEnum()) {
+        return true;
+      }
+    }
+    return false;
+  }
+
+  /**
+   * Ensures that empty string value is allowed for a given type.
+   *
+   * <p>Empty strings are only allowed for String, String Array, Collection of Strings or any of
+   * these types in a parameterized ValueProvider.
+   *
+   * @param type class object for the type under check.
+   * @param genericType complete type information for the type under check.
+   * @param genericTypeName a string representation of the complete type information.
+   */
+  private static void checkEmptyStringAllowed(Class<?> type, JavaType genericType,
+      String genericTypeName) {
+    Class<?> unwrappedType = type.equals(ValueProvider.class)
+        ? genericType.containedType(0).getRawClass() : type;
+
+    Class<?> containedType = unwrappedType;
+    if (unwrappedType.isArray()) {
+      containedType = unwrappedType.getComponentType();
+    } else if (Collection.class.isAssignableFrom(unwrappedType)) {
+      JavaType innerType = type.equals(ValueProvider.class)
+          ? genericType.containedType(0).containedType(0)
+          : genericType.containedType(0);
+      // Note that raw types are allowed, hence the null check.
+      containedType = innerType == null ? String.class : innerType.getRawClass();
+    }
+    if (!containedType.equals(String.class)) {
+      String msg = String.format("Empty argument value is only allowed for String, String Array, "
+              + "Collections of Strings or any of these types in a parameterized ValueProvider, "
+              + "but received: %s",
+          genericTypeName);
+      throw new IllegalArgumentException(msg);
+    }
+  }
+
   @VisibleForTesting
   static Set<String> getSupportedRunners() {
     ImmutableSortedSet.Builder<String> supportedRunners = ImmutableSortedSet.naturalOrder();