[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
@@ -99,6 +100,7 @@
 * (Python) Fixed stateful DoFns with side inputs sometimes taking the timer key coder from a side input instead of the main input, which could make the worker fail to decode timer keys with `Unknown type tag` ([#40374](https://github.com/apache/beam/issues/40374)).
 * (Python) `Duration` built from float seconds now rounds to the nearest microsecond instead of truncating, which could lose a microsecond ([#40263](https://github.com/apache/beam/issues/40263)).
 * (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)).
+* (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 `SplunkIO.Write` now uses a dedicated `SplunkWriteErrorCoder` for its output ([#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/SplunkIO.java

@@ -166,7 +166,8 @@ public PCollection<SplunkWriteError> expand(PCollection<SplunkEvent> input) {
       // Return a PCollection<SplunkWriteError>
       return input
           .apply("Create KV pairs", CreateKeys.of(parallelism()))
-          .apply("Write Splunk events", ParDo.of(writer));
+          .apply("Write Splunk events", ParDo.of(writer))
+          .setCoder(SplunkWriteErrorCoder.of());
     }
 
     /** A builder for creating {@link Write} objects. */

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

@@ -0,0 +1,84 @@
+/*
+ * 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 java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import org.apache.beam.sdk.coders.AtomicCoder;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.coders.NullableCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.values.TypeDescriptor;
+
+/**
+ * A {@link org.apache.beam.sdk.coders.Coder} for {@link SplunkWriteError} objects.
+ *
+ * <p>{@link SplunkWriteError} accessors don't follow the getter naming convention, so the schema
+ * inferred for it has no fields and a schema coder would drop every value.
+ */
+public class SplunkWriteErrorCoder extends AtomicCoder<SplunkWriteError> {
+
+  private static final SplunkWriteErrorCoder SPLUNK_WRITE_ERROR_CODER = new SplunkWriteErrorCoder();
+
+  private static final TypeDescriptor<SplunkWriteError> TYPE_DESCRIPTOR =
+      new TypeDescriptor<SplunkWriteError>() {};
+  private static final NullableCoder<String> STRING_NULLABLE_CODER =
+      NullableCoder.of(StringUtf8Coder.of());
+  private static final NullableCoder<Integer> INTEGER_NULLABLE_CODER =
+      NullableCoder.of(VarIntCoder.of());
+
+  public static SplunkWriteErrorCoder of() {
+    return SPLUNK_WRITE_ERROR_CODER;
+  }
+
+  @Override
+  public void encode(SplunkWriteError value, OutputStream out) throws IOException {
+    INTEGER_NULLABLE_CODER.encode(value.statusCode(), out);
+    STRING_NULLABLE_CODER.encode(value.statusMessage(), out);
+    STRING_NULLABLE_CODER.encode(value.payload(), out);
+  }
+
+  @Override
+  public SplunkWriteError decode(InputStream in) throws CoderException, IOException {
+    SplunkWriteError.Builder builder = SplunkWriteError.newBuilder();
+
+    Integer statusCode = INTEGER_NULLABLE_CODER.decode(in);
+    if (statusCode != null) {
+      builder.withStatusCode(statusCode);
+    }
+
+    String statusMessage = STRING_NULLABLE_CODER.decode(in);
+    if (statusMessage != null) {
+      builder.withStatusMessage(statusMessage);
+    }
+
+    String payload = STRING_NULLABLE_CODER.decode(in);
+    if (payload != null) {
+      builder.withPayload(payload);
+    }
+
+    return builder.create();
+  }
+
+  @Override
+  public TypeDescriptor<SplunkWriteError> getEncodedTypeDescriptor() {
+    return TYPE_DESCRIPTOR;
+  }
+}

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();
+  }
+}