Add http-event-listener.connect-http-headers.config-file property

Introduces the `http-event-listener.connect-http-headers.config-file`
property to allow defining custom HTTP headers in an external file
using `key=value` pairs. This supports header names and values
containing commas (`,`) or colons (`:`), which are not supported in
the inline `http-event-listener.connect-http-headers` property.

Only one of `http-event-listener.connect-http-headers` or
`http-event-listener.connect-http-headers.config-file` may be set.
An exception will be raised if both are configured.

Author: danielbelchior

docs/src/main/sphinx/admin/event-listeners-http.md

@@ -75,6 +75,11 @@ event-listener.config-files=etc/http-event-listener.properties,...
     [](http-event-listener-custom-headers) for more details
   - Empty
 
+* - http-event-listener.connect-http-headers.config-file
+  - Path of the config file containing a list of custom HTTP headers to be sent along with the events. See
+    [](http-event-listener-custom-headers) for more details
+  - Empty
+
 * - http-event-listener.connect-http-method
   - Specifies the HTTP method to use for the request. Supported values
     are POST and PUT.
@@ -119,8 +124,23 @@ Providing headers follows the pattern of `key:value` pairs separated by commas:
 http-event-listener.connect-http-headers="Header-Name-1:header value 1,Header-Value-2:header value 2,..."
 ```
 
-If you need to use a comma(`,`) or colon(`:`) in a header name or value,
-escape it using a backslash (`\`).
+If your header names or values need to include special characters such as commas 
+(`,`) or colons (`:`),define them in an external configuration file using:
+
+```text
+http-event-listener.connect-http-headers.config-file=/path/to/headers.conf
+```
+
+The configuration file should contain one `key=value` pair per line, for example:
+
+```text
+Header-Name-1=header value 1
+Header-Name-2=header value with : colon and , comma
+```
+
+Only one of `http-event-listener.connect-http-headers` **or**
+`http-event-listener.connect-http-headers.config-file` can be used at a time.
+If both properties are set, the system will raise an exception during startup.
 
-Keep in mind that these are static, so they can not carry information
-taken from the event itself.
+Keep in mind that these headers are static—they cannot include information
+dynamically taken from the event itself.

plugin/trino-http-event-listener/pom.xml

@@ -157,6 +157,12 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>io.airlift</groupId>
+            <artifactId>testing</artifactId>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>io.trino</groupId>
             <artifactId>trino-main</artifactId>

plugin/trino-http-event-listener/src/main/java/io/trino/plugin/httpquery/HttpEventListener.java

@@ -19,9 +19,11 @@
 import com.google.common.util.concurrent.Futures;
 import com.google.inject.Inject;
 import io.airlift.bootstrap.LifeCycleManager;
+import io.airlift.configuration.validation.FileExists;
 import io.airlift.http.client.BodyGenerator;
 import io.airlift.http.client.HttpClient;
 import io.airlift.http.client.Request;
+import io.airlift.http.client.StatusResponseHandler.StatusResponse;
 import io.airlift.json.JsonCodec;
 import io.airlift.log.Logger;
 import io.airlift.units.Duration;
@@ -31,9 +33,11 @@
 import io.trino.spi.eventlistener.SplitCompletedEvent;
 import jakarta.annotation.PreDestroy;
 
+import java.io.File;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
@@ -42,7 +46,6 @@
 import static com.google.common.net.MediaType.JSON_UTF_8;
 import static io.airlift.concurrent.Threads.daemonThreadsNamed;
 import static io.airlift.http.client.JsonBodyGenerator.jsonBodyGenerator;
-import static io.airlift.http.client.StatusResponseHandler.StatusResponse;
 import static io.airlift.http.client.StatusResponseHandler.createStatusResponseHandler;
 import static java.util.Objects.requireNonNull;
 import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor;
@@ -73,6 +76,7 @@ public class HttpEventListener
     private final Duration maxDelay;
     private final double backoffBase;
     private final Map<String, String> httpHeaders;
+    private final Optional<@FileExists File> httpHeadersConfigFile;
     private final URI ingestUri;
     private final HttpEventListenerHttpMethod httpMethod;
     private final ScheduledExecutorService executor;
@@ -102,6 +106,7 @@ public HttpEventListener(
         this.backoffBase = config.getBackoffBase();
         this.httpMethod = config.getHttpMethod();
         this.httpHeaders = ImmutableMap.copyOf(config.getHttpHeaders());
+        this.httpHeadersConfigFile = config.getHttpHeadersConfigFile();
 
         try {
             ingestUri = new URI(config.getIngestUri());
@@ -145,13 +150,22 @@ public void splitCompleted(SplitCompletedEvent splitCompletedEvent)
 
     private void sendLog(BodyGenerator eventBodyGenerator, String queryId)
     {
-        Request request = Request.builder()
+        Request.Builder requestBuilder = Request.builder()
                 .setMethod(httpMethod.name())
-                .addHeaders(Multimaps.forMap(httpHeaders))
                 .addHeader(CONTENT_TYPE, JSON_UTF_8.toString())
                 .setUri(ingestUri)
-                .setBodyGenerator(eventBodyGenerator)
-                .build();
+                .setBodyGenerator(eventBodyGenerator);
+
+        httpHeadersConfigFile.ifPresentOrElse(
+                file -> {
+                    Map<String, String> fileHeaders = HttpEventListenerConfig.loadHttpHeadersFromFile(file);
+                    requestBuilder.addHeaders(Multimaps.forMap(fileHeaders));
+                },
+                () -> {
+                    requestBuilder.addHeaders(Multimaps.forMap(httpHeaders));
+                });
+
+        Request request = requestBuilder.build();
 
         attemptToSend(request, 0, Duration.valueOf("0s"), queryId);
     }

plugin/trino-http-event-listener/src/main/java/io/trino/plugin/httpquery/HttpEventListenerConfig.java

@@ -13,18 +13,27 @@
  */
 package io.trino.plugin.httpquery;
 
-import com.google.common.collect.ImmutableMap;
 import io.airlift.configuration.Config;
 import io.airlift.configuration.ConfigDescription;
+import io.airlift.configuration.validation.FileExists;
 import io.airlift.units.Duration;
+import jakarta.validation.constraints.AssertTrue;
 import jakarta.validation.constraints.Min;
 import jakarta.validation.constraints.NotNull;
 
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.util.EnumSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
 import java.util.stream.Collectors;
 
+import static com.google.common.collect.ImmutableMap.toImmutableMap;
+
 public class HttpEventListenerConfig
 {
     private int retryCount;
@@ -34,7 +43,8 @@ public class HttpEventListenerConfig
     private final EnumSet<HttpEventListenerEventType> loggedEvents = EnumSet.noneOf(HttpEventListenerEventType.class);
     private String ingestUri;
     private HttpEventListenerHttpMethod httpMethod = HttpEventListenerHttpMethod.POST;
-    private Map<String, String> httpHeaders = ImmutableMap.of();
+    private Map<String, String> httpHeaders = Map.of();
+    private File httpHeadersConfigFile;
 
     @ConfigDescription("Will log io.trino.spi.eventlistener.QueryCreatedEvent")
     @Config("http-event-listener.log-created")
@@ -130,6 +140,20 @@ public HttpEventListenerConfig setHttpHeaders(List<String> httpHeaders)
         return this;
     }
 
+    public Optional<@FileExists File> getHttpHeadersConfigFile()
+    {
+        return Optional.ofNullable(httpHeadersConfigFile);
+    }
+
+    @Config("http-event-listener.connect-http-headers.config-file")
+    @ConfigDescription("Path to a properties file containing custom HTTP headers, " +
+            "specified as key-value pairs (one per line, e.g., Header-Name=Header Value)")
+    public HttpEventListenerConfig setHttpHeadersConfigFile(File httpHeadersConfigFile)
+    {
+        this.httpHeadersConfigFile = httpHeadersConfigFile;
+        return this;
+    }
+
     @ConfigDescription("Number of retries on server error")
     @Config("http-event-listener.connect-retry-count")
     public HttpEventListenerConfig setRetryCount(int retryCount)
@@ -184,4 +208,27 @@ public Duration getMaxDelay()
     {
         return this.maxDelay;
     }
+
+    @AssertTrue(message = "Exactly one of http-event-listener.connect-http-headers.config-file or " +
+            "http-event-listener.connect-http-headers must be set")
+    public boolean validateHeaderConfigRedundant()
+    {
+        return !(httpHeadersConfigFile != null && !httpHeaders.isEmpty());
+    }
+
+    public static Map<String, String> loadHttpHeadersFromFile(File file)
+    {
+        Properties properties = new Properties();
+        try (FileInputStream fis = new FileInputStream(file)) {
+            properties.load(fis);
+        }
+        catch (IOException e) {
+            throw new UncheckedIOException("Failed to read HTTP headers config file: " + file, e);
+        }
+
+        return properties.entrySet().stream()
+                .collect(toImmutableMap(
+                        e -> e.getKey().toString(),
+                        e -> e.getValue().toString()));
+    }
 }

plugin/trino-http-event-listener/src/test/java/io/trino/plugin/httpquery/TestHttpEventListenerConfig.java

@@ -13,25 +13,81 @@
  */
 package io.trino.plugin.httpquery;
 
+import com.google.common.collect.ImmutableMap;
+import io.airlift.configuration.validation.FileExists;
 import io.airlift.units.Duration;
 import org.junit.jupiter.api.Test;
 
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 
 import static io.airlift.configuration.testing.ConfigAssertions.assertFullMapping;
 import static io.airlift.configuration.testing.ConfigAssertions.assertRecordedDefaults;
 import static io.airlift.configuration.testing.ConfigAssertions.recordDefaults;
+import static io.airlift.testing.ValidationAssertions.assertFailsValidation;
+import static org.assertj.core.api.Assertions.assertThat;
 
 final class TestHttpEventListenerConfig
 {
     @Test
-    void testDefaults()
+    void testGetHttpHeadersLoadsFromConfigFile()
             throws Exception
+    {
+        // Create a temp file with header values in properties format, including single and double quotes
+        Path tempFile = Files.createTempFile("headers", ".properties");
+        String fileContent = "Authorization=Trust Me\nCache-Control=no-cache\nCustom-Header='single-quoted'\nAnother-Header=\"double-quoted\"\n";
+        Files.writeString(tempFile, fileContent);
+
+        HttpEventListenerConfig config = new HttpEventListenerConfig();
+        config.setHttpHeadersConfigFile(tempFile.toFile());
+
+        Map<String, String> headers = config.getHttpHeadersConfigFile()
+                .map(HttpEventListenerConfig::loadHttpHeadersFromFile)
+                .orElseGet(Map::of);
+
+        assertThat(headers)
+                .hasSize(4)
+                .containsEntry("Authorization", "Trust Me")
+                .containsEntry("Cache-Control", "no-cache")
+                .containsEntry("Custom-Header", "'single-quoted'")
+                .containsEntry("Another-Header", "\"double-quoted\"");
+    }
+
+    @Test
+    void testValidateHeaderConfigRedundant()
+            throws IOException
+    {
+        HttpEventListenerConfig config = new HttpEventListenerConfig();
+        // Neither set: valid
+        assertThat(config.validateHeaderConfigRedundant()).isTrue();
+
+        // Only httpHeaders set: valid
+        config.setHttpHeaders(List.of("Authorization: Trust Me"));
+        assertThat(config.validateHeaderConfigRedundant()).isTrue();
+
+        // Only httpHeadersConfigFile set: valid
+        config = new HttpEventListenerConfig();
+        config.setHttpHeadersConfigFile(Files.createTempFile(null, null).toFile());
+        assertThat(config.validateHeaderConfigRedundant()).isTrue();
+
+        // Both set: invalid
+        config.setHttpHeaders(List.of("Authorization: Trust Me"));
+        assertThat(config.validateHeaderConfigRedundant()).isFalse();
+    }
+
+    @Test
+    void testDefaults()
     {
         assertRecordedDefaults(recordDefaults(HttpEventListenerConfig.class)
                 .setHttpHeaders(List.of())
+                .setHttpHeadersConfigFile(null)
                 .setIngestUri(null)
                 .setRetryCount(0)
                 .setRetryDelay(Duration.succinctDuration(1, TimeUnit.SECONDS))
@@ -44,33 +100,78 @@ void testDefaults()
     }
 
     @Test
-    void testExplicitPropertyMappings()
-            throws Exception
+    void testExplicitPropertyMappingsSkippingConnectHttpHeaders()
+            throws IOException
     {
-        Map<String, String> properties = Map.of(
-                "http-event-listener.log-created", "true",
-                "http-event-listener.log-completed", "true",
-                "http-event-listener.log-split", "true",
-                "http-event-listener.connect-ingest-uri", "http://example.com:8080/api",
-                "http-event-listener.connect-http-headers", "Authorization: Trust Me, Cache-Control: no-cache",
-                "http-event-listener.connect-retry-count", "2",
-                "http-event-listener.connect-http-method", "PUT",
-                "http-event-listener.connect-retry-delay", "101s",
-                "http-event-listener.connect-backoff-base", "1.5",
-                "http-event-listener.connect-max-delay", "10m");
+        Path httpHeadersConfigFile = Files.createTempFile(null, null);
+
+        Map<String, String> properties = Map.ofEntries(
+                Map.entry("http-event-listener.connect-http-headers.config-file", httpHeadersConfigFile.toString()),
+                Map.entry("http-event-listener.log-created", "true"),
+                Map.entry("http-event-listener.log-completed", "true"),
+                Map.entry("http-event-listener.log-split", "true"),
+                Map.entry("http-event-listener.connect-ingest-uri", "http://example.com:8080/api"),
+                Map.entry("http-event-listener.connect-retry-count", "2"),
+                Map.entry("http-event-listener.connect-http-method", "PUT"),
+                Map.entry("http-event-listener.connect-retry-delay", "101s"),
+                Map.entry("http-event-listener.connect-backoff-base", "1.5"),
+                Map.entry("http-event-listener.connect-max-delay", "10m"));
 
         HttpEventListenerConfig expected = new HttpEventListenerConfig()
+                .setHttpHeadersConfigFile(httpHeadersConfigFile.toFile())
                 .setLogCompleted(true)
                 .setLogCreated(true)
                 .setLogSplit(true)
                 .setIngestUri("http://example.com:8080/api")
+                .setRetryCount(2)
+                .setHttpMethod(HttpEventListenerHttpMethod.PUT)
+                .setRetryDelay(Duration.succinctDuration(101, TimeUnit.SECONDS))
+                .setBackoffBase(1.5)
+                .setMaxDelay(Duration.succinctDuration(10, TimeUnit.MINUTES));
+
+        assertFullMapping(properties, expected, Set.of("http-event-listener.connect-http-headers"));
+    }
+
+    @Test
+    void testExplicitPropertyMappings()
+    {
+        ImmutableMap<String, String> properties = ImmutableMap.<String, String>builder()
+                .put("http-event-listener.connect-http-headers", "Authorization: Trust Me, Cache-Control: no-cache")
+                .put("http-event-listener.log-created", "true")
+                .put("http-event-listener.log-completed", "true")
+                .put("http-event-listener.log-split", "true")
+                .put("http-event-listener.connect-ingest-uri", "http://example.com:8080/api")
+                .put("http-event-listener.connect-retry-count", "2")
+                .put("http-event-listener.connect-http-method", "PUT")
+                .put("http-event-listener.connect-retry-delay", "101s")
+                .put("http-event-listener.connect-backoff-base", "1.5")
+                .put("http-event-listener.connect-max-delay", "10m")
+                .buildOrThrow();
+
+        HttpEventListenerConfig expected = new HttpEventListenerConfig()
                 .setHttpHeaders(List.of("Authorization: Trust Me", "Cache-Control: no-cache"))
+                .setLogCompleted(true)
+                .setLogCreated(true)
+                .setLogSplit(true)
+                .setIngestUri("http://example.com:8080/api")
                 .setRetryCount(2)
                 .setHttpMethod(HttpEventListenerHttpMethod.PUT)
                 .setRetryDelay(Duration.succinctDuration(101, TimeUnit.SECONDS))
                 .setBackoffBase(1.5)
                 .setMaxDelay(Duration.succinctDuration(10, TimeUnit.MINUTES));
 
-        assertFullMapping(properties, expected);
+        assertFullMapping(properties, expected, Set.of("http-event-listener.connect-http-headers.config-file"));
+    }
+
+    @Test
+    void testConfigFileDoesNotExist()
+    {
+        File file = new File("/doesNotExist-" + UUID.randomUUID());
+        assertFailsValidation(
+                new HttpEventListenerConfig()
+                        .setHttpHeadersConfigFile(file),
+                "httpHeadersConfigFile",
+                "file does not exist: " + file,
+                FileExists.class);
     }
 }