[YAML] Add SplunkIO write SchemaTransform and WriteToSplunk

CHANGES.md

@@ -64,6 +64,7 @@
 
 ## I/Os
 
+* Added a SchemaTransform for SplunkIO writes, available in Beam YAML as `WriteToSplunk` (Java/YAML) ([#40470](https://github.com/apache/beam/issues/40470)).
 * Support for X source added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
 
 ## New Features / Improvements
@@ -109,6 +110,7 @@
 * (Python) `ReadFromText` with `escapechar` no longer skips a delimiter that directly follows an escaped delimiter, which merged two records into one ([#40459](https://github.com/apache/beam/issues/40459)).
 * (Python) DataFrame API: `agg` with a list that includes `'count'` or `'size'` now labels those rows `count`/`size` instead of `sum` ([#40504](https://github.com/apache/beam/issues/40504)).
 * (Python) DataFrame API: `Series.value_counts(bins=...)` no longer always fails with a `TypeError` ([#40503](https://github.com/apache/beam/issues/40503)).
+* (Java) SplunkIO write errors no longer lose their status code, status message and payload when encoded. The schema inferred for `SplunkWriteError` had no fields, so it now has a hand-written schema with `statusCode`, `statusMessage` and `payload` ([#40470](https://github.com/apache/beam/issues/40470)).
 * Fixed X (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
 
 ## Security Fixes

sdks/java/io/expansion-service/build.gradle

@@ -99,6 +99,7 @@ dependencies {
   runtimeOnly project(":sdks:java:io:datadog")
   runtimeOnly project(":sdks:java:io:mongodb")
   runtimeOnly project(":sdks:java:io:delta")
+  runtimeOnly project(":sdks:java:io:splunk")
   runtimeOnly library.java.kafka_clients
   runtimeOnly library.java.slf4j_jdk14
 

sdks/java/io/splunk/src/main/java/org/apache/beam/sdk/io/splunk/SplunkEvent.java

@@ -43,6 +43,14 @@
 @AutoValue
 public abstract class SplunkEvent {
 
+  public static final String TIME = "time";
+  public static final String HOST = "host";
+  public static final String SOURCE = "source";
+  public static final String SOURCE_TYPE = "sourcetype";
+  public static final String INDEX = "index";
+  public static final String FIELDS = "fields";
+  public static final String EVENT = "event";
+
   /** Provides a builder for creating {@link SplunkEvent} objects. */
   public static Builder newBuilder() {
     return new AutoValue_SplunkEvent.Builder();

sdks/java/io/splunk/src/main/java/org/apache/beam/sdk/io/splunk/SplunkWriteError.java

@@ -20,15 +20,14 @@
 import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
 
 import com.google.auto.value.AutoValue;
-import org.apache.beam.sdk.schemas.AutoValueSchema;
 import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
 /**
  * A class for capturing errors that occur while writing {@link SplunkEvent} to Splunk's Http Event
  * Collector (HEC) end point.
  */
-@DefaultSchema(AutoValueSchema.class)
+@DefaultSchema(SplunkWriteErrorSchemaProvider.class)
 @AutoValue
 public abstract class SplunkWriteError {
 

sdks/java/io/splunk/src/main/java/org/apache/beam/sdk/io/splunk/SplunkWriteErrorSchemaProvider.java

@@ -0,0 +1,125 @@
+/*
+ * 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.splunk;
+
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.SchemaProvider;
+import org.apache.beam.sdk.transforms.SerializableFunction;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.sdk.values.TypeDescriptor;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * A {@link SchemaProvider} for {@link SplunkWriteError} objects.
+ *
+ * <p>{@link SplunkWriteError} accessors don't follow the getter naming convention, so the schema
+ * {@link org.apache.beam.sdk.schemas.AutoValueSchema} infers for it has no fields and its schema
+ * coder drops every value. This provider maps the accessors to row fields by hand.
+ */
+public class SplunkWriteErrorSchemaProvider implements SchemaProvider {
+
+  static final String STATUS_CODE = "statusCode";
+  static final String STATUS_MESSAGE = "statusMessage";
+  static final String PAYLOAD = "payload";
+
+  static final Schema SCHEMA =
+      Schema.builder()
+          .addNullableField(STATUS_CODE, Schema.FieldType.INT32)
+          .addNullableField(STATUS_MESSAGE, Schema.FieldType.STRING)
+          .addNullableField(PAYLOAD, Schema.FieldType.STRING)
+          .build();
+
+  @Override
+  public <T> @Nullable Schema schemaFor(TypeDescriptor<T> typeDescriptor) {
+    return SCHEMA;
+  }
+
+  @Override
+  public <T> @Nullable SerializableFunction<T, Row> toRowFunction(
+      TypeDescriptor<T> typeDescriptor) {
+    return new ToRowFn<>();
+  }
+
+  @Override
+  public <T> @Nullable SerializableFunction<Row, T> fromRowFunction(
+      TypeDescriptor<T> typeDescriptor) {
+    return new FromRowFn<>();
+  }
+
+  /** Converts a {@link SplunkWriteError} to a {@link Row} with {@link #SCHEMA}. */
+  private static class ToRowFn<T> implements SerializableFunction<T, Row> {
+    @Override
+    public Row apply(T input) {
+      SplunkWriteError error = (SplunkWriteError) checkNotNull(input);
+      return Row.withSchema(SCHEMA)
+          .addValue(error.statusCode())
+          .addValue(error.statusMessage())
+          .addValue(error.payload())
+          .build();
+    }
+
+    // Equal instances keep schema coders built from this provider equal.
+    @Override
+    public boolean equals(@Nullable Object other) {
+      return other instanceof ToRowFn;
+    }
+
+    @Override
+    public int hashCode() {
+      return ToRowFn.class.hashCode();
+    }
+  }
+
+  /** Converts a {@link Row} with {@link #SCHEMA} back to a {@link SplunkWriteError}. */
+  private static class FromRowFn<T> implements SerializableFunction<Row, T> {
+    @Override
+    @SuppressWarnings("unchecked")
+    public T apply(Row row) {
+      SplunkWriteError.Builder builder = SplunkWriteError.newBuilder();
+
+      Integer statusCode = row.getInt32(STATUS_CODE);
+      if (statusCode != null) {
+        builder.withStatusCode(statusCode);
+      }
+
+      String statusMessage = row.getString(STATUS_MESSAGE);
+      if (statusMessage != null) {
+        builder.withStatusMessage(statusMessage);
+      }
+
+      String payload = row.getString(PAYLOAD);
+      if (payload != null) {
+        builder.withPayload(payload);
+      }
+
+      return (T) builder.create();
+    }
+
+    @Override
+    public boolean equals(@Nullable Object other) {
+      return other instanceof FromRowFn;
+    }
+
+    @Override
+    public int hashCode() {
+      return FromRowFn.class.hashCode();
+    }
+  }
+}

sdks/java/io/splunk/src/main/java/org/apache/beam/sdk/io/splunk/SplunkWriteSchemaTransformConfiguration.java

@@ -0,0 +1,116 @@
+/*
+ * 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.splunk;
+
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument;
+
+import com.google.auto.value.AutoValue;
+import org.apache.beam.sdk.schemas.AutoValueSchema;
+import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
+import org.apache.beam.sdk.schemas.annotations.SchemaFieldDescription;
+import org.apache.beam.sdk.schemas.transforms.providers.ErrorHandling;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Strings;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * Configuration for writing to Splunk's Http Event Collector (HEC).
+ *
+ * <p>This class is meant to be used with {@link SplunkWriteSchemaTransformProvider}.
+ */
+@DefaultSchema(AutoValueSchema.class)
+@AutoValue
+public abstract class SplunkWriteSchemaTransformConfiguration {
+
+  public void validate() {
+    String invalidConfigMessage = "Invalid Splunk Write configuration: ";
+    checkArgument(!getUrl().isEmpty(), invalidConfigMessage + "url must be specified.");
+    checkArgument(!getToken().isEmpty(), invalidConfigMessage + "token must be specified.");
+    Integer batchCount = getBatchCount();
+    if (batchCount != null) {
+      checkArgument(batchCount > 0, invalidConfigMessage + "batchCount must be greater than 0.");
+    }
+    Integer parallelism = getParallelism();
+    if (parallelism != null) {
+      checkArgument(parallelism > 0, invalidConfigMessage + "parallelism must be greater than 0.");
+    }
+    ErrorHandling errorHandling = getErrorHandling();
+    if (errorHandling != null) {
+      checkArgument(
+          !Strings.isNullOrEmpty(errorHandling.getOutput()),
+          invalidConfigMessage + "Output must not be empty if error handling specified.");
+    }
+  }
+
+  /** Instantiates a {@link SplunkWriteSchemaTransformConfiguration.Builder} instance. */
+  public static SplunkWriteSchemaTransformConfiguration.Builder builder() {
+    return new AutoValue_SplunkWriteSchemaTransformConfiguration.Builder();
+  }
+
+  @SchemaFieldDescription("The Splunk HEC endpoint URL, e.g. https://splunk-host:8088.")
+  public abstract String getUrl();
+
+  @SchemaFieldDescription("The Splunk HEC authentication token.")
+  public abstract String getToken();
+
+  @SchemaFieldDescription("The number of events to batch together for each write.")
+  public abstract @Nullable Integer getBatchCount();
+
+  @SchemaFieldDescription("The number of parallel requests to the HEC endpoint.")
+  public abstract @Nullable Integer getParallelism();
+
+  @SchemaFieldDescription(
+      "Whether to disable SSL certificate validation, e.g. for self-signed certificates.")
+  public abstract @Nullable Boolean getDisableCertificateValidation();
+
+  @SchemaFieldDescription("Path to a root CA certificate used to validate the HEC endpoint.")
+  public abstract @Nullable String getRootCaCertificatePath();
+
+  @SchemaFieldDescription("Whether to log the result of each batch write.")
+  public abstract @Nullable Boolean getEnableBatchLogs();
+
+  @SchemaFieldDescription(
+      "Whether requests sent to the HEC endpoint should be GZIP encoded. Defaults to true.")
+  public abstract @Nullable Boolean getEnableGzipHttpCompression();
+
+  @SchemaFieldDescription("Specifies how to handle errors.")
+  public abstract @Nullable ErrorHandling getErrorHandling();
+
+  @AutoValue.Builder
+  public abstract static class Builder {
+    public abstract Builder setUrl(String url);
+
+    public abstract Builder setToken(String token);
+
+    public abstract Builder setBatchCount(Integer batchCount);
+
+    public abstract Builder setParallelism(Integer parallelism);
+
+    public abstract Builder setDisableCertificateValidation(Boolean disableCertificateValidation);
+
+    public abstract Builder setRootCaCertificatePath(String rootCaCertificatePath);
+
+    public abstract Builder setEnableBatchLogs(Boolean enableBatchLogs);
+
+    public abstract Builder setEnableGzipHttpCompression(Boolean enableGzipHttpCompression);
+
+    public abstract Builder setErrorHandling(@Nullable ErrorHandling errorHandling);
+
+    /** Builds the {@link SplunkWriteSchemaTransformConfiguration} configuration. */
+    public abstract SplunkWriteSchemaTransformConfiguration build();
+  }
+}