Add GoogleAdsIO for reading from Google Ads

.test-infra/jenkins/job_PreCommit_Java.groovy

@@ -34,6 +34,7 @@ def excludePaths = [
   'io/elasticsearch',
   'io/elasticsearch-tests',
   'io/file-schema-transform',
+  'io/google-ads',
   'io/google-cloud-platform',
   'io/hadoop-common',
   'io/hadoop-file-system',

.test-infra/jenkins/job_PreCommit_Java_IOs.groovy

@@ -81,6 +81,7 @@ def ioModulesMap = [
     'debezium',
     'elasticsearch',
     'file-schema-transform',
+    'google-ads',
     'hbase',
     'hcatalog',
     'influxdb',

CHANGES.md

@@ -63,6 +63,7 @@
 * Java KafkaIO now supports picking up topics via topicPattern ([#26948](https://github.com/apache/beam/pull/26948))
 * Support for read from Cosmos DB Core SQL API ([#23604](https://github.com/apache/beam/issues/23604))
 * Upgraded to HBase 2.5.5 for HBaseIO. (Java) ([#27711](https://github.com/apache/beam/issues/19554))
+* Added support for GoogleAdsIO source (Java) ([#27681](https://github.com/apache/beam/pull/27681)).
 
 ## New Features / Improvements
 

build.gradle.kts

@@ -335,6 +335,7 @@ tasks.register("javaioPreCommit") {
   dependsOn(":sdks:java:io:elasticsearch-tests:elasticsearch-tests-common:build")
   dependsOn(":sdks:java:io:elasticsearch:build")
   dependsOn(":sdks:java:io:file-schema-transform:build")
+  dependsOn(":sdks:java:io:google-ads:build")
   dependsOn(":sdks:java:io:hbase:build")
   dependsOn(":sdks:java:io:hcatalog:build")
   dependsOn(":sdks:java:io:influxdb:build")

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

@@ -531,6 +531,7 @@ class BeamModulePlugin implements Plugin<Project> {
     def errorprone_version = "2.10.0"
     // Try to keep gax_version consistent with gax-grpc version in google_cloud_platform_libraries_bom
     def gax_version = "2.31.1"
+    def google_ads_version = "26.0.0"
     def google_clients_version = "2.0.0"
     def google_cloud_bigdataoss_version = "2.2.16"
     // Try to keep google_cloud_spanner_version consistent with google_cloud_spanner_bom in google_cloud_platform_libraries_bom
@@ -651,6 +652,8 @@ class BeamModulePlugin implements Plugin<Project> {
         gax_grpc                                    : "com.google.api:gax-grpc", // google_cloud_platform_libraries_bom sets version
         gax_grpc_test                               : "com.google.api:gax-grpc:$gax_version:testlib", // google_cloud_platform_libraries_bom sets version
         gax_httpjson                                : "com.google.api:gax-httpjson", // google_cloud_platform_libraries_bom sets version
+        google_ads                                  : "com.google.api-ads:google-ads:$google_ads_version",
+        google_ads_stubs_v14                        : "com.google.api-ads:google-ads-stubs-v14:$google_ads_version",
         google_api_client                           : "com.google.api-client:google-api-client:$google_clients_version", // for the libraries using $google_clients_version below.
         google_api_client_jackson2                  : "com.google.api-client:google-api-client-jackson2:$google_clients_version",
         google_api_client_java6                     : "com.google.api-client:google-api-client-java6:$google_clients_version",

sdks/java/build-tools/src/main/resources/beam/checkstyle/suppressions.xml

@@ -60,6 +60,9 @@
   <suppress id="ForbidNonVendoredGrpcProtobuf" files=".*sdk.*extensions.*protobuf.*" />
   <suppress id="ForbidNonVendoredGrpcProtobuf" files=".*sdk.*extensions.*ml.*" />
   <suppress id="ForbidNonVendoredGrpcProtobuf" files=".*sdk.*io.*gcp.*" />
+  <suppress id="ForbidNonVendoredGrpcProtobuf" files=".*sdk.*io.*googleads.*DummyRateLimitPolicy\.java" />
+  <suppress id="ForbidNonVendoredGrpcProtobuf" files=".*sdk.*io.*googleads.*GoogleAdsV14\.java" />
+  <suppress id="ForbidNonVendoredGrpcProtobuf" files=".*sdk.*io.*googleads.*GoogleAdsV14Test\.java" />
   <suppress id="ForbidNonVendoredGrpcProtobuf" files=".*google.*cloud.*spanner.*FakeBatchTransactionId\.java" />
   <suppress id="ForbidNonVendoredGrpcProtobuf" files=".*google.*cloud.*spanner.*FakePartitionFactory\.java" />
   <suppress id="ForbidNonVendoredGrpcProtobuf" files=".*extensions.*sql.*datastore.*" />

sdks/java/io/google-ads/build.gradle

@@ -0,0 +1,45 @@
+/*
+* 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 'org.apache.beam.module' }
+applyJavaNature( automaticModuleName: 'org.apache.beam.sdk.io.googleads')
+
+description = "Apache Beam :: SDKs :: Java :: IO :: Google Ads"
+ext.summary = "IO to read from Google Ads"
+
+dependencies {
+  implementation project(path: ":sdks:java:core", configuration: "shadow")
+  implementation project(path: ":sdks:java:extensions:google-cloud-platform-core")
+  implementation library.java.jackson_annotations
+  implementation library.java.gax
+  implementation library.java.google_ads
+  implementation library.java.google_auth_library_credentials
+  implementation library.java.google_auth_library_oauth2_http
+  implementation library.java.protobuf_java
+  implementation library.java.protobuf_java_util
+  implementation library.java.google_ads
+  implementation library.java.google_ads_stubs_v14
+  implementation library.java.joda_time
+  implementation library.java.vendored_guava_32_1_2_jre
+  testImplementation project(path: ":sdks:java:core", configuration: "shadowTest")
+  testImplementation project(path: ":sdks:java:io:common", configuration: "testRuntimeMigration")
+  testImplementation library.java.mockito_core
+  testImplementation library.java.junit
+  testRuntimeOnly project(path: ":runners:direct-java", configuration: "shadow")
+  testRuntimeOnly library.java.slf4j_jdk14
+}

sdks/java/io/google-ads/src/main/java/org/apache/beam/sdk/io/googleads/DefaultGoogleAdsClientFactory.java

@@ -0,0 +1,67 @@
+/*
+ * 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.googleads;
+
+import com.google.ads.googleads.lib.GoogleAdsClient;
+import com.google.auth.Credentials;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/** The default way to construct a {@link GoogleAdsClient}. */
+public class DefaultGoogleAdsClientFactory implements GoogleAdsClientFactory {
+  private static final DefaultGoogleAdsClientFactory INSTANCE = new DefaultGoogleAdsClientFactory();
+
+  public static DefaultGoogleAdsClientFactory getInstance() {
+    return INSTANCE;
+  }
+
+  @Override
+  public GoogleAdsClient newGoogleAdsClient(
+      GoogleAdsOptions options,
+      @Nullable String developerToken,
+      @Nullable Long linkedCustomerId,
+      @Nullable Long loginCustomerId) {
+
+    GoogleAdsClient.Builder builder = GoogleAdsClient.newBuilder();
+
+    Credentials credentials = options.getGoogleAdsCredential();
+    if (credentials != null) {
+      builder.setCredentials(credentials);
+    }
+
+    if (options.getGoogleAdsEndpoint() != null) {
+      builder.setEndpoint(options.getGoogleAdsEndpoint());
+    }
+
+    String developerTokenFromOptions = options.getGoogleAdsDeveloperToken();
+    if (developerToken != null) {
+      builder.setDeveloperToken(developerToken);
+    } else if (developerTokenFromOptions != null) {
+      builder.setDeveloperToken(developerTokenFromOptions);
+    }
+
+    if (linkedCustomerId != null) {
+      builder.setLinkedCustomerId(linkedCustomerId);
+    }
+
+    if (loginCustomerId != null) {
+      builder.setLoginCustomerId(loginCustomerId);
+    }
+
+    return builder.build();
+  }
+}

sdks/java/io/google-ads/src/main/java/org/apache/beam/sdk/io/googleads/GoogleAdsClientFactory.java

@@ -0,0 +1,31 @@
+/*
+ * 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.googleads;
+
+import com.google.ads.googleads.lib.GoogleAdsClient;
+import java.io.Serializable;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/** Defines how to construct a {@link GoogleAdsClient}. */
+public interface GoogleAdsClientFactory extends Serializable {
+  GoogleAdsClient newGoogleAdsClient(
+      GoogleAdsOptions options,
+      @Nullable String developerToken,
+      @Nullable Long linkedCustomerId,
+      @Nullable Long loginCustomerId);
+}

sdks/java/io/google-ads/src/main/java/org/apache/beam/sdk/io/googleads/GoogleAdsIO.java

@@ -0,0 +1,33 @@
+/*
+ * 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.googleads;
+
+/**
+ * {@link GoogleAdsIO} provides an API for reading from the <a
+ * href="https://developers.google.com/google-ads/api/docs/start">Google Ads API</a> over different
+ * versions of the Google Ads client libraries.
+ *
+ * @see GoogleAdsV14
+ */
+public class GoogleAdsIO {
+  private GoogleAdsIO() {}
+
+  public static GoogleAdsV14 v14() {
+    return GoogleAdsV14.INSTANCE;
+  }
+}

sdks/java/io/google-ads/src/main/java/org/apache/beam/sdk/io/googleads/GoogleAdsOptions.java

@@ -0,0 +1,132 @@
+/*
+ * 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.googleads;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.google.auth.Credentials;
+import java.io.IOException;
+import java.security.GeneralSecurityException;
+import org.apache.beam.sdk.extensions.gcp.auth.CredentialFactory;
+import org.apache.beam.sdk.options.Default;
+import org.apache.beam.sdk.options.DefaultValueFactory;
+import org.apache.beam.sdk.options.Description;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.util.InstanceBuilder;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/** Options used to configure Google Ads API specific options. */
+public interface GoogleAdsOptions extends PipelineOptions {
+  /** Host endpoint to use for connections to the Google Ads API. */
+  @Description("Host endpoint to use for connections to the Google Ads API.")
+  @Default.String("googleads.googleapis.com:443")
+  String getGoogleAdsEndpoint();
+
+  void setGoogleAdsEndpoint(String endpoint);
+
+  /**
+   * OAuth 2.0 Client ID identifying the application.
+   *
+   * @see https://developers.google.com/google-ads/api/docs/oauth/overview
+   * @see https://developers.google.com/identity/protocols/oauth2
+   */
+  @Description("OAuth 2.0 Client ID identifying the application.")
+  String getGoogleAdsClientId();
+
+  void setGoogleAdsClientId(String clientId);
+
+  /**
+   * OAuth 2.0 Client Secret for the specified Client ID.
+   *
+   * @see https://developers.google.com/google-ads/api/docs/oauth/overview
+   * @see https://developers.google.com/identity/protocols/oauth2
+   */
+  @Description("OAuth 2.0 Client Secret for the specified Client ID.")
+  String getGoogleAdsClientSecret();
+
+  void setGoogleAdsClientSecret(String clientSecret);
+
+  /**
+   * OAuth 2.0 Refresh Token for the user connecting to the Google Ads API.
+   *
+   * @see https://developers.google.com/google-ads/api/docs/oauth/overview
+   * @see https://developers.google.com/identity/protocols/oauth2
+   */
+  @Description("OAuth 2.0 Refresh Token for the user connecting to the Google Ads API.")
+  String getGoogleAdsRefreshToken();
+
+  void setGoogleAdsRefreshToken(String refreshToken);
+
+  /** Google Ads developer token for the user connecting to the Google Ads API. */
+  @Description("Google Ads developer token for the user connecting to the Google Ads API.")
+  @Nullable
+  String getGoogleAdsDeveloperToken();
+
+  void setGoogleAdsDeveloperToken(String developerToken);
+
+  /**
+   * The class of the credential factory to create credentials if none have been explicitly set.
+   *
+   * @see #getGoogleAdsCredential()
+   */
+  @Description(
+      "The class of the credential factory to create credentials if none have been explicitly set.")
+  @Default.Class(GoogleAdsUserCredentialFactory.class)
+  Class<? extends CredentialFactory> getGoogleAdsCredentialFactoryClass();
+
+  void setGoogleAdsCredentialFactoryClass(
+      Class<? extends CredentialFactory> credentialFactoryClass);
+
+  /**
+   * The credential instance that should be used to authenticate against the Google Ads API.
+   * Defaults to a credential instance constructed by the credential factory.
+   *
+   * @see #getGoogleAdsCredential()
+   * @see https://github.com/googleapis/google-auth-library-java
+   */
+  @JsonIgnore
+  @Description(
+      "The credential instance that should be used to authenticate against the Google Ads API. "
+          + "Defaults to a credential instance constructed by the credential factory.")
+  @Default.InstanceFactory(GoogleAdsCredentialsFactory.class)
+  @Nullable
+  Credentials getGoogleAdsCredential();
+
+  void setGoogleAdsCredential(Credentials credential);
+
+  /**
+   * Attempts to load the Google Ads credentials. See {@link CredentialFactory#getCredential()} for
+   * more details.
+   */
+  class GoogleAdsCredentialsFactory implements DefaultValueFactory<@Nullable Credentials> {
+    @Override
+    public @Nullable Credentials create(PipelineOptions options) {
+      GoogleAdsOptions googleAdsOptions = options.as(GoogleAdsOptions.class);
+      try {
+        CredentialFactory factory =
+            InstanceBuilder.ofType(CredentialFactory.class)
+                .fromClass(googleAdsOptions.getGoogleAdsCredentialFactoryClass())
+                .fromFactoryMethod("fromOptions")
+                .withArg(PipelineOptions.class, options)
+                .build();
+        return factory.getCredential();
+      } catch (IOException | GeneralSecurityException e) {
+        throw new RuntimeException("Unable to obtain credential", e);
+      }
+    }
+  }
+}