feat(managedkafka): Connect code samples

managedkafka/examples/pom.xml

@@ -3,7 +3,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.example.managedkafka</groupId>
     <artifactId>managedkafka-snippets</artifactId>
-    <packaging>pom</packaging>
+    <packaging>jar</packaging>
     <name>Google Cloud Managed Kafka Snippets</name>
     <url>https://github.com/GoogleCloudPlatform/java-docs-samples/tree/main/managedkafka</url>
 
@@ -29,7 +29,7 @@
             <dependency>
                 <groupId>com.google.cloud</groupId>
                 <artifactId>libraries-bom</artifactId>
-                <version>26.50.0</version>
+                <version>26.64.0</version>
                 <type>pom</type>
                 <scope>import</scope>
             </dependency>

managedkafka/examples/src/main/java/examples/CreateBigQuerySinkConnector.java

@@ -0,0 +1,115 @@
+/*
+ * Copyright 2025 Google LLC
+ *
+ * Licensed 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 examples;
+
+// [START managedkafka_create_bigquery_sink_connector]
+
+import com.google.api.gax.rpc.ApiException;
+import com.google.cloud.managedkafka.v1.ConnectClusterName;
+import com.google.cloud.managedkafka.v1.Connector;
+import com.google.cloud.managedkafka.v1.ConnectorName;
+import com.google.cloud.managedkafka.v1.CreateConnectorRequest;
+import com.google.cloud.managedkafka.v1.ManagedKafkaConnectClient;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+public class CreateBigQuerySinkConnector {
+
+  public static void main(String[] args) throws Exception {
+    // TODO(developer): Replace these variables before running the example.
+    String projectId = "my-project-id";
+    String region = "my-region"; // e.g. us-east1
+    String connectClusterId = "my-connect-cluster";
+    String connectorId = "my-bigquery-sink-connector";
+    String bigqueryProjectId = "my-bigquery-project-id";
+    String datasetName = "my_dataset";
+    String tableName = "kafka_sink_table";
+    String kafkaTopicName = "kafka-topic";
+    String maxTasks = "3";
+    String connectorClass = "com.wepay.kafka.connect.bigquery.BigQuerySinkConnector";
+    String keyConverter = "org.apache.kafka.connect.storage.StringConverter";
+    String valueConverter = "org.apache.kafka.connect.json.JsonConverter";
+    String valueSchemasEnable = "false";
+    createBigQuerySinkConnector(
+        projectId,
+        region,
+        connectClusterId,
+        connectorId,
+        bigqueryProjectId,
+        datasetName,
+        tableName,
+        kafkaTopicName,
+        maxTasks,
+        connectorClass,
+        keyConverter,
+        valueConverter,
+        valueSchemasEnable);
+  }
+
+  public static void createBigQuerySinkConnector(
+      String projectId,
+      String region,
+      String connectClusterId,
+      String connectorId,
+      String bigqueryProjectId,
+      String datasetName,
+      String tableName,
+      String kafkaTopicName,
+      String maxTasks,
+      String connectorClass,
+      String keyConverter,
+      String valueConverter,
+      String valueSchemasEnable)
+      throws Exception {
+
+    // Build the connector configuration
+    Map<String, String> configMap = new HashMap<>();
+    configMap.put("name", connectorId);
+    configMap.put("project", bigqueryProjectId);
+    configMap.put("topics", kafkaTopicName);
+    configMap.put("tasks.max", maxTasks);
+    configMap.put("connector.class", connectorClass);
+    configMap.put("key.converter", keyConverter);
+    configMap.put("value.converter", valueConverter);
+    configMap.put("value.converter.schemas.enable", valueSchemasEnable);
+    configMap.put("defaultDataset", datasetName);
+
+    Connector connector =
+        Connector.newBuilder()
+            .setName(ConnectorName.of(projectId, region, connectClusterId, connectorId).toString())
+            .putAllConfigs(configMap)
+            .build();
+
+    try (ManagedKafkaConnectClient managedKafkaConnectClient = ManagedKafkaConnectClient.create()) {
+      CreateConnectorRequest request =
+          CreateConnectorRequest.newBuilder()
+              .setParent(ConnectClusterName.of(projectId, region, connectClusterId).toString())
+              .setConnectorId(connectorId)
+              .setConnector(connector)
+              .build();
+
+      // This operation is being handled synchronously.
+      Connector response = managedKafkaConnectClient.createConnector(request);
+      System.out.printf("Created BigQuery Sink connector: %s\n", response.getName());
+    } catch (IOException | ApiException e) {
+      System.err.printf("managedKafkaConnectClient.createConnector got err: %s", e.getMessage());
+    }
+  }
+}
+
+// [END managedkafka_create_bigquery_sink_connector]

managedkafka/examples/src/main/java/examples/CreateCloudStorageSinkConnector.java

@@ -0,0 +1,115 @@
+/*
+ * Copyright 2025 Google LLC
+ *
+ * Licensed 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 examples;
+
+// [START managedkafka_create_cloud_storage_sink_connector]
+
+import com.google.api.gax.rpc.ApiException;
+import com.google.cloud.managedkafka.v1.ConnectClusterName;
+import com.google.cloud.managedkafka.v1.Connector;
+import com.google.cloud.managedkafka.v1.ConnectorName;
+import com.google.cloud.managedkafka.v1.CreateConnectorRequest;
+import com.google.cloud.managedkafka.v1.ManagedKafkaConnectClient;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+public class CreateCloudStorageSinkConnector {
+
+  public static void main(String[] args) throws Exception {
+    // TODO(developer): Replace these variables before running the example.
+    String projectId = "my-project-id";
+    String region = "my-region"; // e.g. us-east1
+    String connectClusterId = "my-connect-cluster";
+    String connectorId = "my-gcs-sink-connector";
+    String bucketName = "my-gcs-bucket";
+    String kafkaTopicName = "kafka-topic";
+    String connectorClass = "io.aiven.kafka.connect.gcs.GcsSinkConnector";
+    String maxTasks = "1";
+    String gcsCredentialsDefault = "true";
+    String formatOutputType = "json";
+    String valueConverter = "org.apache.kafka.connect.json.JsonConverter";
+    String valueSchemasEnable = "false";
+    String keyConverter = "org.apache.kafka.connect.storage.StringConverter";
+    createCloudStorageSinkConnector(
+        projectId,
+        region,
+        connectClusterId,
+        connectorId,
+        bucketName,
+        kafkaTopicName,
+        connectorClass,
+        maxTasks,
+        gcsCredentialsDefault,
+        formatOutputType,
+        valueConverter,
+        valueSchemasEnable,
+        keyConverter);
+  }
+
+  public static void createCloudStorageSinkConnector(
+      String projectId,
+      String region,
+      String connectClusterId,
+      String connectorId,
+      String bucketName,
+      String kafkaTopicName,
+      String connectorClass,
+      String maxTasks,
+      String gcsCredentialsDefault,
+      String formatOutputType,
+      String valueConverter,
+      String valueSchemasEnable,
+      String keyConverter)
+      throws Exception {
+
+    // Build the connector configuration
+    Map<String, String> configMap = new HashMap<>();
+    configMap.put("connector.class", connectorClass);
+    configMap.put("tasks.max", maxTasks);
+    configMap.put("topics", kafkaTopicName);
+    configMap.put("gcs.bucket.name", bucketName);
+    configMap.put("gcs.credentials.default", gcsCredentialsDefault);
+    configMap.put("format.output.type", formatOutputType);
+    configMap.put("name", connectorId);
+    configMap.put("value.converter", valueConverter);
+    configMap.put("value.converter.schemas.enable", valueSchemasEnable);
+    configMap.put("key.converter", keyConverter);
+
+    Connector connector = Connector.newBuilder()
+        .setName(
+            ConnectorName.of(projectId, region, connectClusterId, connectorId).toString())
+        .putAllConfigs(configMap)
+        .build();
+
+    try (ManagedKafkaConnectClient managedKafkaConnectClient = ManagedKafkaConnectClient.create()) {
+      CreateConnectorRequest request = CreateConnectorRequest.newBuilder()
+          .setParent(ConnectClusterName.of(projectId, region, connectClusterId).toString())
+          .setConnectorId(connectorId)
+          .setConnector(connector)
+          .build();
+
+      // This operation is being handled synchronously.
+      Connector response = managedKafkaConnectClient.createConnector(request);
+      System.out.printf("Created Cloud Storage Sink connector: %s\n", response.getName());
+    } catch (IOException | ApiException e) {
+      System.err.printf("managedKafkaConnectClient.createConnector got err: %s", e.getMessage());
+    }
+  }
+}
+
+// [END managedkafka_create_cloud_storage_sink_connector]

managedkafka/examples/src/main/java/examples/CreateConnectCluster.java

@@ -0,0 +1,126 @@
+/*
+ * Copyright 2025 Google LLC
+ *
+ * Licensed 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 examples;
+
+// [START managedkafka_create_connect_cluster]
+
+import com.google.api.gax.longrunning.OperationFuture;
+import com.google.api.gax.longrunning.OperationSnapshot;
+import com.google.api.gax.longrunning.OperationTimedPollAlgorithm;
+import com.google.api.gax.retrying.RetrySettings;
+import com.google.api.gax.retrying.RetryingFuture;
+import com.google.api.gax.retrying.TimedRetryAlgorithm;
+import com.google.cloud.managedkafka.v1.CapacityConfig;
+import com.google.cloud.managedkafka.v1.ConnectAccessConfig;
+import com.google.cloud.managedkafka.v1.ConnectCluster;
+import com.google.cloud.managedkafka.v1.ConnectGcpConfig;
+import com.google.cloud.managedkafka.v1.ConnectNetworkConfig;
+import com.google.cloud.managedkafka.v1.CreateConnectClusterRequest;
+import com.google.cloud.managedkafka.v1.LocationName;
+import com.google.cloud.managedkafka.v1.ManagedKafkaConnectClient;
+import com.google.cloud.managedkafka.v1.ManagedKafkaConnectSettings;
+import com.google.cloud.managedkafka.v1.OperationMetadata;
+import java.time.Duration;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+
+public class CreateConnectCluster {
+
+  public static void main(String[] args) throws Exception {
+    // TODO(developer): Replace these variables before running the example.
+    String projectId = "my-project-id";
+    String region = "my-region"; // e.g. us-east1
+    String clusterId = "my-connect-cluster";
+    String subnet = "my-subnet"; // e.g. projects/my-project/regions/my-region/subnetworks/my-subnet
+    String kafkaCluster = "my-kafka-cluster"; // The Kafka cluster to connect to
+    int cpu = 12;
+    long memoryBytes = 12884901888L; // 12 GiB
+    createConnectCluster(projectId, region, clusterId, subnet, kafkaCluster, cpu, memoryBytes);
+  }
+
+  public static void createConnectCluster(
+      String projectId,
+      String region,
+      String clusterId,
+      String subnet,
+      String kafkaCluster,
+      int cpu,
+      long memoryBytes)
+      throws Exception {
+    CapacityConfig capacityConfig = CapacityConfig.newBuilder().setVcpuCount(cpu)
+        .setMemoryBytes(memoryBytes).build();
+    ConnectNetworkConfig networkConfig = ConnectNetworkConfig.newBuilder()
+        .setPrimarySubnet(subnet)
+        .build();
+    ConnectGcpConfig gcpConfig = ConnectGcpConfig.newBuilder()
+        .setAccessConfig(ConnectAccessConfig.newBuilder().addNetworkConfigs(networkConfig).build())
+        .build();
+    ConnectCluster connectCluster = ConnectCluster.newBuilder()
+        .setCapacityConfig(capacityConfig)
+        .setGcpConfig(gcpConfig)
+        .setKafkaCluster(kafkaCluster)
+        .build();
+
+    // Create the settings to configure the timeout for polling operations
+    ManagedKafkaConnectSettings.Builder settingsBuilder = ManagedKafkaConnectSettings.newBuilder();
+    TimedRetryAlgorithm timedRetryAlgorithm = OperationTimedPollAlgorithm.create(
+        RetrySettings.newBuilder()
+            .setTotalTimeoutDuration(Duration.ofHours(1L))
+            .build());
+    settingsBuilder.createConnectClusterOperationSettings()
+        .setPollingAlgorithm(timedRetryAlgorithm);
+
+    try (ManagedKafkaConnectClient managedKafkaConnectClient = ManagedKafkaConnectClient
+        .create(settingsBuilder.build())) {
+      CreateConnectClusterRequest request = CreateConnectClusterRequest.newBuilder()
+          .setParent(LocationName.of(projectId, region).toString())
+          .setConnectClusterId(clusterId)
+          .setConnectCluster(connectCluster)
+          .build();
+
+      // The duration of this operation can vary considerably, typically taking
+      // between 10-30 minutes.
+      OperationFuture<ConnectCluster, OperationMetadata> future = managedKafkaConnectClient
+          .createConnectClusterOperationCallable().futureCall(request);
+
+      // Get the initial LRO and print details.
+      OperationSnapshot operation = future.getInitialFuture().get();
+      System.out.printf(
+          "Connect cluster creation started. Operation name: %s\nDone: %s\nMetadata: %s\n",
+          operation.getName(), operation.isDone(), future.getMetadata().get().toString());
+
+      while (!future.isDone()) {
+        // The pollingFuture gives us the most recent status of the operation
+        RetryingFuture<OperationSnapshot> pollingFuture = future.getPollingFuture();
+        OperationSnapshot currentOp = pollingFuture.getAttemptResult().get();
+        System.out.printf("Polling Operation:\nName: %s\n Done: %s\n",
+            currentOp.getName(),
+            currentOp.isDone());
+      }
+
+      // NOTE: future.get() blocks completion until the operation is complete (isDone
+      // = True)
+      ConnectCluster response = future.get();
+      System.out.printf("Created connect cluster: %s\n", response.getName());
+    } catch (ExecutionException e) {
+      System.err.printf("managedKafkaConnectClient.createConnectCluster got err: %s", 
+          e.getMessage());
+      throw e;
+    }
+  }
+}
+// [END managedkafka_create_connect_cluster]