Add buffer configuration to ClickHouse destination

.generator/schemas/v2/openapi.yaml

@@ -64398,6 +64398,8 @@ components:
           $ref: "#/components/schemas/ObservabilityPipelineClickhouseDestinationBatch"
         batch_encoding:
           $ref: "#/components/schemas/ObservabilityPipelineClickhouseDestinationBatchEncoding"
+        buffer:
+          $ref: "#/components/schemas/ObservabilityPipelineBufferOptions"
         compression:
           $ref: "#/components/schemas/ObservabilityPipelineClickhouseDestinationCompression"
         database:

examples/v2/observability-pipelines/ValidatePipeline_253109967.java

@@ -0,0 +1,151 @@
+// Validate an observability pipeline with ClickHouse destination with all fields set returns "OK"
+// response
+
+import com.datadog.api.client.ApiClient;
+import com.datadog.api.client.ApiException;
+import com.datadog.api.client.v2.api.ObservabilityPipelinesApi;
+import com.datadog.api.client.v2.model.ObservabilityPipelineBufferOptions;
+import com.datadog.api.client.v2.model.ObservabilityPipelineBufferOptionsMemoryType;
+import com.datadog.api.client.v2.model.ObservabilityPipelineBufferOptionsWhenFull;
+import com.datadog.api.client.v2.model.ObservabilityPipelineClickhouseDestination;
+import com.datadog.api.client.v2.model.ObservabilityPipelineClickhouseDestinationAuth;
+import com.datadog.api.client.v2.model.ObservabilityPipelineClickhouseDestinationAuthStrategy;
+import com.datadog.api.client.v2.model.ObservabilityPipelineClickhouseDestinationBatch;
+import com.datadog.api.client.v2.model.ObservabilityPipelineClickhouseDestinationBatchEncoding;
+import com.datadog.api.client.v2.model.ObservabilityPipelineClickhouseDestinationBatchEncodingCodec;
+import com.datadog.api.client.v2.model.ObservabilityPipelineClickhouseDestinationCompression;
+import com.datadog.api.client.v2.model.ObservabilityPipelineClickhouseDestinationCompressionAlgorithm;
+import com.datadog.api.client.v2.model.ObservabilityPipelineClickhouseDestinationCompressionObject;
+import com.datadog.api.client.v2.model.ObservabilityPipelineClickhouseDestinationFormat;
+import com.datadog.api.client.v2.model.ObservabilityPipelineClickhouseDestinationType;
+import com.datadog.api.client.v2.model.ObservabilityPipelineConfig;
+import com.datadog.api.client.v2.model.ObservabilityPipelineConfigDestinationItem;
+import com.datadog.api.client.v2.model.ObservabilityPipelineConfigProcessorGroup;
+import com.datadog.api.client.v2.model.ObservabilityPipelineConfigProcessorItem;
+import com.datadog.api.client.v2.model.ObservabilityPipelineConfigSourceItem;
+import com.datadog.api.client.v2.model.ObservabilityPipelineDataAttributes;
+import com.datadog.api.client.v2.model.ObservabilityPipelineDatadogAgentSource;
+import com.datadog.api.client.v2.model.ObservabilityPipelineDatadogAgentSourceType;
+import com.datadog.api.client.v2.model.ObservabilityPipelineFilterProcessor;
+import com.datadog.api.client.v2.model.ObservabilityPipelineFilterProcessorType;
+import com.datadog.api.client.v2.model.ObservabilityPipelineMemoryBufferSizeOptions;
+import com.datadog.api.client.v2.model.ObservabilityPipelineSpec;
+import com.datadog.api.client.v2.model.ObservabilityPipelineSpecData;
+import com.datadog.api.client.v2.model.ObservabilityPipelineTls;
+import com.datadog.api.client.v2.model.ValidationResponse;
+import java.util.Collections;
+
+public class Example {
+  public static void main(String[] args) {
+    ApiClient defaultClient = ApiClient.getDefaultApiClient();
+    ObservabilityPipelinesApi apiInstance = new ObservabilityPipelinesApi(defaultClient);
+
+    ObservabilityPipelineSpec body =
+        new ObservabilityPipelineSpec()
+            .data(
+                new ObservabilityPipelineSpecData()
+                    .attributes(
+                        new ObservabilityPipelineDataAttributes()
+                            .config(
+                                new ObservabilityPipelineConfig()
+                                    .destinations(
+                                        Collections.singletonList(
+                                            new ObservabilityPipelineConfigDestinationItem(
+                                                new ObservabilityPipelineClickhouseDestination()
+                                                    .id("clickhouse-destination")
+                                                    .inputs(
+                                                        Collections.singletonList(
+                                                            "my-processor-group"))
+                                                    .type(
+                                                        ObservabilityPipelineClickhouseDestinationType
+                                                            .CLICKHOUSE)
+                                                    .endpointUrlKey("CLICKHOUSE_ENDPOINT_URL")
+                                                    .database("my_database")
+                                                    .table("application_logs")
+                                                    .format(
+                                                        ObservabilityPipelineClickhouseDestinationFormat
+                                                            .ARROW_STREAM)
+                                                    .skipUnknownFields(true)
+                                                    .dateTimeBestEffort(true)
+                                                    .compression(
+                                                        new ObservabilityPipelineClickhouseDestinationCompression(
+                                                            new ObservabilityPipelineClickhouseDestinationCompressionObject()
+                                                                .algorithm(
+                                                                    ObservabilityPipelineClickhouseDestinationCompressionAlgorithm
+                                                                        .GZIP)
+                                                                .level(6L)))
+                                                    .auth(
+                                                        new ObservabilityPipelineClickhouseDestinationAuth()
+                                                            .strategy(
+                                                                ObservabilityPipelineClickhouseDestinationAuthStrategy
+                                                                    .BASIC)
+                                                            .usernameKey("CLICKHOUSE_USERNAME")
+                                                            .passwordKey("CLICKHOUSE_PASSWORD"))
+                                                    .batch(
+                                                        new ObservabilityPipelineClickhouseDestinationBatch()
+                                                            .maxEvents(1000L)
+                                                            .timeoutSecs(1L))
+                                                    .batchEncoding(
+                                                        new ObservabilityPipelineClickhouseDestinationBatchEncoding()
+                                                            .codec(
+                                                                ObservabilityPipelineClickhouseDestinationBatchEncodingCodec
+                                                                    .ARROW_STREAM)
+                                                            .allowNullableFields(true))
+                                                    .tls(
+                                                        new ObservabilityPipelineTls()
+                                                            .crtFile("/path/to/cert.crt")
+                                                            .caFile("/path/to/ca.crt")
+                                                            .keyFile("/path/to/key.key")
+                                                            .keyPassKey("TLS_KEY_PASSPHRASE"))
+                                                    .buffer(
+                                                        new ObservabilityPipelineBufferOptions(
+                                                            new ObservabilityPipelineMemoryBufferSizeOptions()
+                                                                .type(
+                                                                    ObservabilityPipelineBufferOptionsMemoryType
+                                                                        .MEMORY)
+                                                                .maxEvents(500L)
+                                                                .whenFull(
+                                                                    ObservabilityPipelineBufferOptionsWhenFull
+                                                                        .BLOCK))))))
+                                    .processorGroups(
+                                        Collections.singletonList(
+                                            new ObservabilityPipelineConfigProcessorGroup()
+                                                .enabled(true)
+                                                .id("my-processor-group")
+                                                .include("service:my-service")
+                                                .inputs(
+                                                    Collections.singletonList(
+                                                        "datadog-agent-source"))
+                                                .processors(
+                                                    Collections.singletonList(
+                                                        new ObservabilityPipelineConfigProcessorItem(
+                                                            new ObservabilityPipelineFilterProcessor()
+                                                                .enabled(true)
+                                                                .id("filter-processor")
+                                                                .include("status:error")
+                                                                .type(
+                                                                    ObservabilityPipelineFilterProcessorType
+                                                                        .FILTER))))))
+                                    .sources(
+                                        Collections.singletonList(
+                                            new ObservabilityPipelineConfigSourceItem(
+                                                new ObservabilityPipelineDatadogAgentSource()
+                                                    .id("datadog-agent-source")
+                                                    .type(
+                                                        ObservabilityPipelineDatadogAgentSourceType
+                                                            .DATADOG_AGENT)))))
+                            .name("Pipeline with ClickHouse Destination All Fields"))
+                    .type("pipelines"));
+
+    try {
+      ValidationResponse result = apiInstance.validatePipeline(body);
+      System.out.println(result);
+    } catch (ApiException e) {
+      System.err.println("Exception when calling ObservabilityPipelinesApi#validatePipeline");
+      System.err.println("Status code: " + e.getCode());
+      System.err.println("Reason: " + e.getResponseBody());
+      System.err.println("Response headers: " + e.getResponseHeaders());
+      e.printStackTrace();
+    }
+  }
+}

src/main/java/com/datadog/api/client/v2/model/ObservabilityPipelineClickhouseDestination.java

@@ -30,6 +30,7 @@
   ObservabilityPipelineClickhouseDestination.JSON_PROPERTY_AUTH,
   ObservabilityPipelineClickhouseDestination.JSON_PROPERTY_BATCH,
   ObservabilityPipelineClickhouseDestination.JSON_PROPERTY_BATCH_ENCODING,
+  ObservabilityPipelineClickhouseDestination.JSON_PROPERTY_BUFFER,
   ObservabilityPipelineClickhouseDestination.JSON_PROPERTY_COMPRESSION,
   ObservabilityPipelineClickhouseDestination.JSON_PROPERTY_DATABASE,
   ObservabilityPipelineClickhouseDestination.JSON_PROPERTY_DATE_TIME_BEST_EFFORT,
@@ -55,6 +56,9 @@ public class ObservabilityPipelineClickhouseDestination {
   public static final String JSON_PROPERTY_BATCH_ENCODING = "batch_encoding";
   private ObservabilityPipelineClickhouseDestinationBatchEncoding batchEncoding;
 
+  public static final String JSON_PROPERTY_BUFFER = "buffer";
+  private ObservabilityPipelineBufferOptions buffer;
+
   public static final String JSON_PROPERTY_COMPRESSION = "compression";
   private ObservabilityPipelineClickhouseDestinationCompression compression;
 
@@ -179,6 +183,29 @@ public void setBatchEncoding(
     this.batchEncoding = batchEncoding;
   }
 
+  public ObservabilityPipelineClickhouseDestination buffer(
+      ObservabilityPipelineBufferOptions buffer) {
+    this.buffer = buffer;
+    this.unparsed |= buffer.unparsed;
+    return this;
+  }
+
+  /**
+   * Configuration for buffer settings on destination components.
+   *
+   * @return buffer
+   */
+  @jakarta.annotation.Nullable
+  @JsonProperty(JSON_PROPERTY_BUFFER)
+  @JsonInclude(value = JsonInclude.Include.USE_DEFAULTS)
+  public ObservabilityPipelineBufferOptions getBuffer() {
+    return buffer;
+  }
+
+  public void setBuffer(ObservabilityPipelineBufferOptions buffer) {
+    this.buffer = buffer;
+  }
+
   public ObservabilityPipelineClickhouseDestination compression(
       ObservabilityPipelineClickhouseDestinationCompression compression) {
     this.compression = compression;
@@ -509,6 +536,7 @@ public boolean equals(Object o) {
         && Objects.equals(this.batch, observabilityPipelineClickhouseDestination.batch)
         && Objects.equals(
             this.batchEncoding, observabilityPipelineClickhouseDestination.batchEncoding)
+        && Objects.equals(this.buffer, observabilityPipelineClickhouseDestination.buffer)
         && Objects.equals(this.compression, observabilityPipelineClickhouseDestination.compression)
         && Objects.equals(this.database, observabilityPipelineClickhouseDestination.database)
         && Objects.equals(
@@ -534,6 +562,7 @@ public int hashCode() {
         auth,
         batch,
         batchEncoding,
+        buffer,
         compression,
         database,
         dateTimeBestEffort,
@@ -555,6 +584,7 @@ public String toString() {
     sb.append("    auth: ").append(toIndentedString(auth)).append("\n");
     sb.append("    batch: ").append(toIndentedString(batch)).append("\n");
     sb.append("    batchEncoding: ").append(toIndentedString(batchEncoding)).append("\n");
+    sb.append("    buffer: ").append(toIndentedString(buffer)).append("\n");
     sb.append("    compression: ").append(toIndentedString(compression)).append("\n");
     sb.append("    database: ").append(toIndentedString(database)).append("\n");
     sb.append("    dateTimeBestEffort: ").append(toIndentedString(dateTimeBestEffort)).append("\n");

src/test/resources/cassettes/features/v2/Validate_an_observability_pipeline_with_ClickHouse_destination_with_all_fields_set_returns_OK_response.freeze

@@ -0,0 +1 @@
+2026-06-24T16:45:05.037Z

src/test/resources/cassettes/features/v2/Validate_an_observability_pipeline_with_ClickHouse_destination_with_all_fields_set_returns_OK_response.json

@@ -0,0 +1,32 @@
+[
+  {
+    "httpRequest": {
+      "body": {
+        "type": "JSON",
+        "json": "{\"data\":{\"attributes\":{\"config\":{\"destinations\":[{\"auth\":{\"password_key\":\"CLICKHOUSE_PASSWORD\",\"strategy\":\"basic\",\"username_key\":\"CLICKHOUSE_USERNAME\"},\"batch\":{\"max_events\":1000,\"timeout_secs\":1},\"batch_encoding\":{\"allow_nullable_fields\":true,\"codec\":\"arrow_stream\"},\"buffer\":{\"max_events\":500,\"type\":\"memory\",\"when_full\":\"block\"},\"compression\":{\"algorithm\":\"gzip\",\"level\":6},\"database\":\"my_database\",\"date_time_best_effort\":true,\"endpoint_url_key\":\"CLICKHOUSE_ENDPOINT_URL\",\"format\":\"arrow_stream\",\"id\":\"clickhouse-destination\",\"inputs\":[\"my-processor-group\"],\"skip_unknown_fields\":true,\"table\":\"application_logs\",\"tls\":{\"ca_file\":\"/path/to/ca.crt\",\"crt_file\":\"/path/to/cert.crt\",\"key_file\":\"/path/to/key.key\",\"key_pass_key\":\"TLS_KEY_PASSPHRASE\"},\"type\":\"clickhouse\"}],\"processor_groups\":[{\"enabled\":true,\"id\":\"my-processor-group\",\"include\":\"service:my-service\",\"inputs\":[\"datadog-agent-source\"],\"processors\":[{\"enabled\":true,\"id\":\"filter-processor\",\"include\":\"status:error\",\"type\":\"filter\"}]}],\"sources\":[{\"id\":\"datadog-agent-source\",\"type\":\"datadog_agent\"}]},\"name\":\"Pipeline with ClickHouse Destination All Fields\"},\"type\":\"pipelines\"}}"
+      },
+      "headers": {},
+      "method": "POST",
+      "path": "/api/v2/obs-pipelines/pipelines/validate",
+      "keepAlive": false,
+      "secure": true
+    },
+    "httpResponse": {
+      "body": "{\"errors\":[]}\n",
+      "headers": {
+        "Content-Type": [
+          "application/vnd.api+json"
+        ]
+      },
+      "statusCode": 200,
+      "reasonPhrase": "OK"
+    },
+    "times": {
+      "remainingTimes": 1
+    },
+    "timeToLive": {
+      "unlimited": true
+    },
+    "id": "b5389bbc-bd4c-392a-8c9d-a4f2d324e4d8"
+  }
+]

src/test/resources/com/datadog/api/client/v2/api/observability_pipelines.feature

@@ -192,6 +192,14 @@ Feature: Observability Pipelines
     Then the response status is 200 OK
     And the response "errors" has length 0
 
+  @team:DataDog/observability-pipelines
+  Scenario: Validate an observability pipeline with ClickHouse destination with all fields set returns "OK" response
+    Given new "ValidatePipeline" request
+    And body with value {"data": {"attributes": {"config": {"destinations": [{"id": "clickhouse-destination", "inputs": ["my-processor-group"], "type": "clickhouse", "endpoint_url_key": "CLICKHOUSE_ENDPOINT_URL", "database": "my_database", "table": "application_logs", "format": "arrow_stream", "skip_unknown_fields": true, "date_time_best_effort": true, "compression": {"algorithm": "gzip", "level": 6}, "auth": {"strategy": "basic", "username_key": "CLICKHOUSE_USERNAME", "password_key": "CLICKHOUSE_PASSWORD"}, "batch": {"max_events": 1000, "timeout_secs": 1}, "batch_encoding": {"codec": "arrow_stream", "allow_nullable_fields": true}, "tls": {"crt_file": "/path/to/cert.crt", "ca_file": "/path/to/ca.crt", "key_file": "/path/to/key.key", "key_pass_key": "TLS_KEY_PASSPHRASE"}, "buffer": {"type": "memory", "max_events": 500, "when_full": "block"}}], "processor_groups": [{"enabled": true, "id": "my-processor-group", "include": "service:my-service", "inputs": ["datadog-agent-source"], "processors": [{"enabled": true, "id": "filter-processor", "include": "status:error", "type": "filter"}]}], "sources": [{"id": "datadog-agent-source", "type": "datadog_agent"}]}, "name": "Pipeline with ClickHouse Destination All Fields"}, "type": "pipelines"}}
+    When the request is sent
+    Then the response status is 200 OK
+    And the response "errors" has length 0
+
   @team:DataDog/observability-pipelines
   Scenario: Validate an observability pipeline with HTTP server source valid_tokens returns "OK" response
     Given new "ValidatePipeline" request