[BEAM-14048] [CdapIO] Add ConfigWrapper for building CDAP PluginConfigs

buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy

@@ -451,6 +451,7 @@ class BeamModulePlugin implements Plugin<Project> {
     def aws_java_sdk_version = "1.12.135"
     def aws_java_sdk2_version = "2.17.127"
     def cassandra_driver_version = "3.10.2"
+    def cdap_version = "6.5.1"
     def checkerframework_version = "3.10.0"
     def classgraph_version = "4.8.104"
     def errorprone_version = "2.10.0"
@@ -535,6 +536,8 @@ class BeamModulePlugin implements Plugin<Project> {
         bigdataoss_util                             : "com.google.cloud.bigdataoss:util:$google_cloud_bigdataoss_version",
         cassandra_driver_core                       : "com.datastax.cassandra:cassandra-driver-core:$cassandra_driver_version",
         cassandra_driver_mapping                    : "com.datastax.cassandra:cassandra-driver-mapping:$cassandra_driver_version",
+        cdap_api                                    : "io.cdap.cdap:cdap-api:$cdap_version",
+        cdap_etl_api                                : "io.cdap.cdap:cdap-etl-api:$cdap_version",
         checker_qual                                : "org.checkerframework:checker-qual:$checkerframework_version",
         classgraph                                  : "io.github.classgraph:classgraph:$classgraph_version",
         commons_codec                               : "commons-codec:commons-codec:1.15",

sdks/java/io/cdap/OWNERS

@@ -0,0 +1 @@
+# See the OWNERS docs at https://s.apache.org/beam-owners

sdks/java/io/cdap/build.gradle

@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * License); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an AS IS BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+plugins {
+    id 'java'
+    id 'org.apache.beam.module'
+}
+
+applyJavaNature(
+        exportJavadoc: false,
+        automaticModuleName: 'org.apache.beam.sdk.io.cdap',
+)
+provideIntegrationTestingDependencies()
+enableJavaPerformanceTesting()
+
+description = "Apache Beam :: CDAP :: Java"
+ext.summary = """Apache Beam SDK provides a simple, Java-based
+interface for integration with CDAP plugins."""
+
+/** Define the list of runners which execute a precommit test.
+ * Some runners are run from separate projects, see the preCommit task below
+ * for details.
+ */
+
+allprojects {
+    repositories {
+        maven { url 'https://jitpack.io' }
+    }
+}
+
+dependencies {
+    implementation library.java.cdap_api
+    implementation library.java.jackson_core
+    implementation library.java.jackson_databind
+    implementation library.java.slf4j_api
+    implementation project(path: ":sdks:java:core", configuration: "shadow")
+    testImplementation "com.github.data-integrations:salesforce:v1.3.9"
+    testImplementation library.java.cdap_etl_api
+    testImplementation library.java.vendored_guava_26_0_jre
+    testImplementation library.java.junit
+}

sdks/java/io/cdap/src/main/java/org/apache/beam/sdk/io/cdap/ConfigWrapper.java

@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.cdap;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.cdap.cdap.api.plugin.PluginConfig;
+import java.io.File;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Class for building {@link PluginConfig} object of the specific class {@param <T>}. */
+public class ConfigWrapper<T extends PluginConfig> {
+
+  private static final Logger LOG = LoggerFactory.getLogger(ConfigWrapper.class);
+
+  @Nullable private Map<String, Object> paramsMap = null;
+  private final Class<T> configClass;
+
+  public ConfigWrapper(Class<T> configClass) {
+    this.configClass = configClass;
+  }
+
+  public ConfigWrapper<T> fromJsonString(String jsonString) throws IOException {
+    TypeReference<HashMap<String, Object>> typeRef =
+        new TypeReference<HashMap<String, Object>>() {};
+    try {
+      paramsMap = new ObjectMapper().readValue(jsonString, typeRef);
+    } catch (IOException e) {
+      LOG.error("Can not read json string to params map", e);
+      throw e;
+    }
+    return this;
+  }
+
+  public ConfigWrapper<T> fromJsonFile(File jsonFile) throws IOException {
+    TypeReference<HashMap<String, Object>> typeRef =
+        new TypeReference<HashMap<String, Object>>() {};
+    try {
+      paramsMap = new ObjectMapper().readValue(jsonFile, typeRef);
+    } catch (IOException e) {
+      LOG.error("Can not read json file to params map", e);
+      throw e;
+    }
+    return this;
+  }
+
+  public ConfigWrapper<T> withParams(Map<String, Object> paramsMap) {
+    this.paramsMap = new HashMap<>(paramsMap);
+    return this;
+  }
+
+  public ConfigWrapper<T> setParam(String paramName, Object param) {
+    getParamsMap().put(paramName, param);
+    return this;
+  }
+
+  public @Nullable T build() {
+    return PluginConfigInstantiationUtils.getPluginConfig(getParamsMap(), configClass);
+  }
+
+  private @Nonnull Map<String, Object> getParamsMap() {
+    if (paramsMap == null) {
+      paramsMap = new HashMap<>();
+    }
+    return paramsMap;
+  }
+}

sdks/java/io/cdap/src/main/java/org/apache/beam/sdk/io/cdap/PluginConfigInstantiationUtils.java

@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.cdap;
+
+import io.cdap.cdap.api.annotation.Name;
+import io.cdap.cdap.api.plugin.PluginConfig;
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Modifier;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Class for getting any filled {@link PluginConfig} configuration object. */
+@SuppressWarnings({"unchecked", "assignment.type.incompatible"})
+public class PluginConfigInstantiationUtils {
+
+  private static final Logger LOG = LoggerFactory.getLogger(PluginConfigInstantiationUtils.class);
+
+  /**
+   * Method for instantiating {@link PluginConfig} object of specific class {@param configClass}.
+   * After instantiating, it will go over all {@link Field}s with the {@link Name} annotation and
+   * set the appropriate parameter values from the {@param params} map for them.
+   *
+   * @param params map of config fields, where key is the name of the field, value must be String or
+   *     boxed primitive
+   * @return Config object for given map of arguments and configuration class
+   */
+  static @Nullable <T extends PluginConfig> T getPluginConfig(
+      Map<String, Object> params, Class<T> configClass) {
+    // Validate configClass
+    if (configClass == null) {
+      throw new IllegalArgumentException("Config class must be not null!");
+    }
+    List<Field> allFields = new ArrayList<>();
+    Class<?> currClass = configClass;
+    while (currClass != null && !currClass.equals(Object.class)) {
+      allFields.addAll(
+          Arrays.stream(currClass.getDeclaredFields())
+              .filter(
+                  f -> !Modifier.isStatic(f.getModifiers()) && f.isAnnotationPresent(Name.class))
+              .collect(Collectors.toList()));
+      currClass = currClass.getSuperclass();
+    }
+    T config = getEmptyObjectFromDefaultValues(configClass);
+
+    if (config != null) {
+      for (Field field : allFields) {
+        field.setAccessible(true);
+
+        Class<?> fieldType = field.getType();
+
+        Name declaredAnnotation = field.getDeclaredAnnotation(Name.class);
+        Object fieldValue =
+            declaredAnnotation != null ? params.get(declaredAnnotation.value()) : null;
+
+        if (fieldValue != null && fieldType.equals(fieldValue.getClass())) {
+          try {
+            field.set(config, fieldValue);
+          } catch (IllegalAccessException e) {
+            LOG.error("Can not set a field with value {}", fieldValue);
+          }
+        }
+      }
+    }
+    return config;
+  }
+
+  /** @return empty {@link Object} of {@param tClass} */
+  private static @Nullable <T> T getEmptyObjectFromDefaultValues(Class<T> tClass) {
+    for (Constructor<?> constructor : tClass.getDeclaredConstructors()) {
+      constructor.setAccessible(true);
+      Class<?>[] parameterTypes = constructor.getParameterTypes();
+      Object[] parameters = new Object[parameterTypes.length];
+      for (int i = 0; i < parameterTypes.length; i++) {
+        parameters[i] = getDefaultValue(parameterTypes[i]);
+      }
+      try {
+        return (T) constructor.newInstance(parameters);
+      } catch (InstantiationException | IllegalAccessException | InvocationTargetException e) {
+        LOG.warn("Can not instantiate an empty object", e);
+      }
+    }
+    return null;
+  }
+
+  /** @return default value for given {@param tClass} if it's primitive, otherwise returns null */
+  private static @Nullable Object getDefaultValue(@Nullable Class<?> tClass) {
+    if (Boolean.TYPE.equals(tClass)) {
+      return false;
+    }
+    if (Character.TYPE.equals(tClass)) {
+      return Character.MIN_VALUE;
+    }
+    if (Byte.TYPE.equals(tClass)) {
+      return Byte.MIN_VALUE;
+    }
+    if (Short.TYPE.equals(tClass)) {
+      return Short.MIN_VALUE;
+    }
+    if (Double.TYPE.equals(tClass)) {
+      return Double.MIN_VALUE;
+    }
+    if (Integer.TYPE.equals(tClass)) {
+      return Integer.MIN_VALUE;
+    }
+    if (Float.TYPE.equals(tClass)) {
+      return Float.MIN_VALUE;
+    }
+    if (Long.TYPE.equals(tClass)) {
+      return Long.MIN_VALUE;
+    }
+    return null;
+  }
+}

sdks/java/io/cdap/src/main/java/org/apache/beam/sdk/io/cdap/package-info.java

@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/** Transforms for reading and writing from CDAP. */
+@Experimental(Kind.SOURCE_SINK)
+package org.apache.beam.sdk.io.cdap;
+
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;