[CdapIO] Complete examples for CDAP Zendesk plugins (#24589)

* Add examples for Cdap Zendesk plugins

* Move common classes to Examples Cdap module

* Fix readme

examples/java/cdap/README.md

@@ -24,4 +24,4 @@ Supported CDAP plugins:
 - [ServiceNow](https://github.com/data-integrations/servicenow-plugins). More info in the ServiceNow example [README](servicenow/src/main/java/org/apache/beam/examples/complete/cdap/servicenow/README.md).
 - [Salesforce](https://github.com/data-integrations/salesforce)
 - [Hubspot](https://github.com/data-integrations/hubspot)
-- [Zendesk](https://github.com/data-integrations/zendesk)
+- [Zendesk](https://github.com/data-integrations/zendesk). More info in the ServiceNow example [README](zendesk/src/main/java/org/apache/beam/examples/complete/cdap/zendesk/README.md).

examples/java/cdap/zendesk/build.gradle

@@ -0,0 +1,106 @@
+/*
+ * 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.
+ */
+
+import groovy.json.JsonOutput
+
+plugins {
+    id 'java'
+    id 'org.apache.beam.module'
+    id 'com.github.johnrengelman.shadow'
+}
+
+applyJavaNature(
+        exportJavadoc: false,
+        automaticModuleName: 'org.apache.beam.examples.complete.cdap.zendesk',
+)
+
+description = "Apache Beam :: Examples :: Java :: CDAP :: Zendesk"
+ext.summary = """Apache Beam SDK provides a simple, Java-based
+interface for processing virtually any size data. This
+artifact includes CDAP Zendesk Apache Beam Java SDK examples."""
+
+/** Define the list of runners which execute a precommit test.
+ * Some runners are run from separate projects, see the preCommit task below
+ * for details.
+ */
+def preCommitRunners = ["directRunner", "flinkRunner"]
+for (String runner : preCommitRunners) {
+    configurations.create(runner + "PreCommit")
+}
+
+dependencies {
+    implementation enforcedPlatform(library.java.google_cloud_platform_libraries_bom)
+    implementation project(path: ":sdks:java:core", configuration: "shadow")
+    implementation project(":examples:java:cdap")
+    implementation project(":sdks:java:io:cdap")
+    implementation project(":sdks:java:io:hadoop-common")
+    implementation library.java.cdap_api
+    implementation library.java.cdap_api_commons
+    implementation library.java.cdap_etl_api
+    permitUnusedDeclared library.java.cdap_etl_api
+    implementation library.java.cdap_hydrator_common
+    implementation library.java.cdap_plugin_zendesk
+    implementation library.java.hadoop_common
+    implementation library.java.slf4j_api
+    implementation library.java.vendored_guava_26_0_jre
+    runtimeOnly project(path: ":runners:direct-java", configuration: "shadow")
+
+    // Add dependencies for the PreCommit configurations
+    // For each runner a project level dependency on the examples project.
+    for (String runner : preCommitRunners) {
+        delegate.add(runner + "PreCommit", project(":examples:java:cdap:zendesk"))
+        delegate.add(runner + "PreCommit", project(path: ":examples:java:cdap:zendesk", configuration: "testRuntimeMigration"))
+    }
+    directRunnerPreCommit project(path: ":runners:direct-java", configuration: "shadow")
+    flinkRunnerPreCommit project(":runners:flink:${project.ext.latestFlinkVersion}")
+}
+
+/*
+ * Create a ${runner}PreCommit task for each runner which runs a set
+ * of integration tests for WordCount and WindowedWordCount.
+ */
+def preCommitRunnerClass = [
+        directRunner: "org.apache.beam.runners.direct.DirectRunner",
+        flinkRunner: "org.apache.beam.runners.flink.TestFlinkRunner"
+]
+
+for (String runner : preCommitRunners) {
+    tasks.create(name: runner + "PreCommit", type: Test) {
+        def preCommitBeamTestPipelineOptions = [
+                "--runner=" + preCommitRunnerClass[runner],
+        ]
+        classpath = configurations."${runner}PreCommit"
+        forkEvery 1
+        maxParallelForks 4
+        systemProperty "beamTestPipelineOptions", JsonOutput.toJson(preCommitBeamTestPipelineOptions)
+    }
+}
+
+/* Define a common precommit task which depends on all the individual precommits. */
+task preCommit() {
+    for (String runner : preCommitRunners) {
+        dependsOn runner + "PreCommit"
+    }
+}
+
+task executeCdap (type:JavaExec) {
+    mainClass = System.getProperty("mainClass")
+    classpath = sourceSets.main.runtimeClasspath
+    systemProperties System.getProperties()
+    args System.getProperty("exec.args", "").split()
+}

examples/java/cdap/zendesk/src/main/java/org/apache/beam/examples/complete/cdap/zendesk/CdapZendeskToTxt.java

@@ -0,0 +1,140 @@
+/*
+ * 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.examples.complete.cdap.zendesk;
+
+import static org.apache.beam.examples.complete.cdap.zendesk.transforms.FormatInputTransform.readFromCdapZendesk;
+
+import io.cdap.cdap.api.data.format.StructuredRecord;
+import java.util.Map;
+import org.apache.beam.examples.complete.cdap.utils.StructuredRecordUtils;
+import org.apache.beam.examples.complete.cdap.zendesk.options.CdapZendeskOptions;
+import org.apache.beam.examples.complete.cdap.zendesk.utils.PluginConfigOptionsConverter;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.PipelineResult;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.NullableCoder;
+import org.apache.beam.sdk.coders.SerializableCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.io.TextIO;
+import org.apache.beam.sdk.io.hadoop.WritableCoder;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.transforms.MapValues;
+import org.apache.beam.sdk.transforms.Values;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.apache.hadoop.io.NullWritable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * The {@link CdapZendeskToTxt} pipeline is a batch pipeline which ingests data in JSON format from
+ * CDAP Zendesk, and outputs the resulting records to .txt file. Zendesk parameters and output txt
+ * file path are specified by the user as template parameters. <br>
+ *
+ * <p><b>Example Usage</b>
+ *
+ * <pre>
+ * # Gradle preparation
+ *
+ * To run this example your {@code build.gradle} file should contain the following task
+ * to execute the pipeline:
+ * {@code
+ * task executeCdap (type:JavaExec) {
+ *     mainClass = System.getProperty("mainClass")
+ *     classpath = sourceSets.main.runtimeClasspath
+ *     systemProperties System.getProperties()
+ *     args System.getProperty("exec.args", "").split()
+ * }
+ * }
+ *
+ * This task allows to run the pipeline via the following command:
+ * {@code
+ * gradle clean executeCdap -DmainClass=org.apache.beam.examples.complete.cdap.zendesk.CdapZendeskToTxt \
+ *      -Dexec.args="--<argument>=<value> --<argument>=<value>"
+ * }
+ *
+ * # Running the pipeline
+ * To execute this pipeline, specify the parameters in the following format:
+ * {@code
+ * --zendeskBaseUrl=your-url \
+ * --adminEmail=your-email \
+ * --apiToken=your-token \
+ * --objectsToPull=Groups \
+ * --referenceName=your-reference-name \
+ * --outputTxtFilePathPrefix=your-path-to-output-folder-with-filename-prefix
+ * }
+ *
+ * By default this will run the pipeline locally with the DirectRunner. To change the runner, specify:
+ * {@code
+ * --runner=YOUR_SELECTED_RUNNER
+ * }
+ * </pre>
+ */
+public class CdapZendeskToTxt {
+
+  /* Logger for class.*/
+  private static final Logger LOG = LoggerFactory.getLogger(CdapZendeskToTxt.class);
+
+  /**
+   * Main entry point for pipeline execution.
+   *
+   * @param args Command line arguments to the pipeline.
+   */
+  public static void main(String[] args) {
+    CdapZendeskOptions options =
+        PipelineOptionsFactory.fromArgs(args).withValidation().as(CdapZendeskOptions.class);
+
+    // Create the pipeline
+    Pipeline pipeline = Pipeline.create(options);
+    run(pipeline, options);
+  }
+
+  /**
+   * Runs a pipeline which reads message from CDAP and writes it to .txt file.
+   *
+   * @param options arguments to the pipeline
+   */
+  public static PipelineResult run(Pipeline pipeline, CdapZendeskOptions options) {
+    Map<String, Object> pluginConfigParams =
+        PluginConfigOptionsConverter.zendeskOptionsToParamsMap(options);
+    LOG.info("Starting Cdap-Zendesk-To-Txt pipeline with parameters: {}", pluginConfigParams);
+
+    /*
+     * Steps:
+     *  1) Read messages in from Cdap Zendesk
+     *  2) Extract values only
+     *  3) Write successful records to .txt file
+     */
+
+    pipeline
+        .apply("readFromCdapZendesk", readFromCdapZendesk(pluginConfigParams))
+        .setCoder(
+            KvCoder.of(
+                NullableCoder.of(WritableCoder.of(NullWritable.class)),
+                SerializableCoder.of(StructuredRecord.class)))
+        .apply(
+            MapValues.into(TypeDescriptors.strings())
+                .via(StructuredRecordUtils::structuredRecordToString))
+        .setCoder(
+            KvCoder.of(
+                NullableCoder.of(WritableCoder.of(NullWritable.class)), StringUtf8Coder.of()))
+        .apply(Values.create())
+        .apply("writeToTxt", TextIO.write().to(options.getOutputTxtFilePathPrefix()));
+
+    return pipeline.run();
+  }
+}

examples/java/cdap/zendesk/src/main/java/org/apache/beam/examples/complete/cdap/zendesk/README.md

@@ -0,0 +1,55 @@
+<!--
+    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.
+-->
+
+## Gradle preparation
+
+To run this example your `build.gradle` file should contain the following task to execute the pipeline:
+
+```
+task executeCdap (type:JavaExec) {
+    mainClass = System.getProperty("mainClass")
+    classpath = sourceSets.main.runtimeClasspath
+    systemProperties System.getProperties()
+    args System.getProperty("exec.args", "").split()
+}
+```
+
+## Running the CdapZendeskToTxt pipeline example
+
+Gradle 'executeCdap' task allows to run the pipeline via the following command:
+
+```bash
+gradle clean executeCdap -DmainClass=org.apache.beam.examples.complete.cdap.zendesk.CdapZendeskToTxt \
+     -Dexec.args="--<argument>=<value> --<argument>=<value>"
+```
+
+To execute this pipeline, specify the parameters in the following format:
+
+```bash
+ --zendeskBaseUrl=zendesk-url-key-followed-by-/%s/%s (example: https://support.zendesk.com/%s/%s) \
+ --adminEmail=your-admin-admin-email \
+ --apiToken=your-api-token \
+ --subdomains=your-subdomains (example: api/v2) \
+ --maxRetryCount=your-max-retry-count \
+ --maxRetryWait=your-max-retry-wait \
+ --maxRetryJitterWait=your-max-retry-jitter-wait \
+ --connectTimeout=your-connection-timeout \
+ --readTimeout=your-read-timeout \
+ --objectsToPull=your-objects-to-pull (example: Groups) \
+ --outputTxtFilePathPrefix=your-path-to-output-folder-with-filename-prefix
+```
+Please see CDAP [Zendesk Batch Source](https://github.com/data-integrations/zendesk/blob/develop/docs/Zendesk-batchsource.md) for more information.