CASSANALYTICS-177: Add instanceId query parameter to fix HTTP 421 err…

CHANGES.txt

@@ -1,5 +1,6 @@
 0.5.0
 -----
+ * Add sidecar.instance.id Spark conf to append an instanceId query parameter to outbound sidecar requests, fixing 421 errors when Sidecar is behind a load balancer (CASSANALYTICS-177)
  * Upgrade sidecar version to 0.4.0
  * Exclude IP address from RingInstance equality so node replacement does not fail bulk write jobs (CASSANALYTICS-175)
  * Regenerate bloom filters for CQLSSTableWriter (CASSANALYTICS-167)

analytics-sidecar-client-common/src/main/java/org/apache/cassandra/sidecar/common/http/SidecarQueryParamNames.java

@@ -0,0 +1,40 @@
+/*
+ * 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.cassandra.sidecar.common.http;
+
+/**
+ * Custom query parameter names for sidecar HTTP requests.
+ */
+public final class SidecarQueryParamNames
+{
+    /**
+     * {@code "instanceId"} query parameter. When present on an outbound sidecar request it carries
+     * the job-level instance identifier supplied by the client (see the Spark conf key
+     * {@code spark.cassandra_analytics.sidecar.instance.id}).
+     *
+     * <p>Requires a Sidecar server &gt;= 0.2.0 (see {@code AbstractHandler#host}, introduced in
+     * CASSSIDECAR-208); older servers do not resolve this parameter and requests will fall back to
+     * Host-header-based instance resolution.
+     */
+    public static final String INSTANCE_ID = "instanceId";
+
+    private SidecarQueryParamNames()
+    {
+    }
+}

analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/HttpClientConfig.java

@@ -38,6 +38,7 @@ public class HttpClientConfig
     public static final String DEFAULT_TRUST_STORE_TYPE = "JKS";
     public static final String DEFAULT_KEY_STORE_TYPE = "PKCS12";
     public static final String DEFAULT_CASSANDRA_ROLE = null;
+    public static final Integer DEFAULT_INSTANCE_ID = null;
 
     private final long timeoutMillis;
     private final boolean ssl;
@@ -54,6 +55,7 @@ public class HttpClientConfig
     private final String keyStorePassword;
     private final String keyStoreType;
     private final String cassandraRole;
+    private final Integer instanceId;
 
     private HttpClientConfig(Builder<?> builder)
     {
@@ -72,6 +74,7 @@ private HttpClientConfig(Builder<?> builder)
         keyStorePassword = builder.keyStorePassword;
         keyStoreType = builder.keyStoreType;
         cassandraRole = builder.cassandraRole;
+        instanceId = builder.instanceId;
     }
 
     /**
@@ -192,6 +195,15 @@ public String cassandraRole()
         return cassandraRole;
     }
 
+    /**
+     * @return the job-level sidecar instance identifier, or {@code null} to omit the {@code instanceId} query parameter
+     */
+    @Nullable
+    public Integer instanceId()
+    {
+        return instanceId;
+    }
+
     /**
      * {@code HttpClient} builder static inner class.
      *
@@ -214,6 +226,7 @@ public static class Builder<T extends Builder<T>>
         private String keyStorePassword;
         private String keyStoreType = DEFAULT_KEY_STORE_TYPE;
         private String cassandraRole = DEFAULT_CASSANDRA_ROLE;
+        private Integer instanceId = DEFAULT_INSTANCE_ID;
 
         /**
          * @return a reference to itself
@@ -412,6 +425,26 @@ public T cassandraRole(String cassandraRole)
             return self();
         }
 
+        /**
+         * Sets the {@code instanceId} query parameter appended to every outbound sidecar request,
+         * and returns a reference to this Builder enabling method chaining. Non-null values must
+         * be greater than or equal to {@code 0}.
+         *
+         * @param instanceId the {@code instanceId} to set, or {@code null} to disable it
+         * @return a reference to this Builder
+         */
+        public T instanceId(Integer instanceId)
+        {
+            // Re-validated in BulkSparkConf.getSidecarInstanceId() to surface a Spark-conf-specific
+            // error message early; keep this constraint (>= 0) in sync with that check.
+            if (instanceId != null && instanceId < 0)
+            {
+                throw new IllegalArgumentException("instanceId must be greater than or equal to 0");
+            }
+            this.instanceId = instanceId;
+            return self();
+        }
+
         /**
          * Returns a {@code SidecarClientConfig} built from the parameters previously set.
          *

analytics-sidecar-client/src/test/java/org/apache/cassandra/sidecar/client/HttpClientConfigTest.java

@@ -23,6 +23,7 @@
 import org.junit.jupiter.api.Test;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.Mockito.mock;
 
 /**
@@ -159,4 +160,40 @@ void testCassandraRole()
         HttpClientConfig config = new HttpClientConfig.Builder<>().cassandraRole("custom_role").build();
         assertThat(config.cassandraRole()).isEqualTo("custom_role");
     }
+
+    @Test
+    void testInstanceIdDefaultIsNull()
+    {
+        HttpClientConfig config = new HttpClientConfig.Builder<>().build();
+        assertThat(config.instanceId()).isNull();
+    }
+
+    @Test
+    void testInstanceId()
+    {
+        HttpClientConfig config = new HttpClientConfig.Builder<>().instanceId(42).build();
+        assertThat(config.instanceId()).isEqualTo(42);
+    }
+
+    @Test
+    void testInstanceIdZeroIsAllowed()
+    {
+        HttpClientConfig config = new HttpClientConfig.Builder<>().instanceId(0).build();
+        assertThat(config.instanceId()).isEqualTo(0);
+    }
+
+    @Test
+    void testInstanceIdNullDisablesIt()
+    {
+        HttpClientConfig config = new HttpClientConfig.Builder<>().instanceId(null).build();
+        assertThat(config.instanceId()).isNull();
+    }
+
+    @Test
+    void testInstanceIdNegativeThrows()
+    {
+        assertThatThrownBy(() -> new HttpClientConfig.Builder<>().instanceId(-1))
+        .isExactlyInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("instanceId must be greater than or equal to 0");
+    }
 }

analytics-sidecar-vertx-client/src/main/java/org/apache/cassandra/sidecar/client/VertxHttpClient.java

@@ -58,6 +58,7 @@
 import org.apache.cassandra.sidecar.common.request.UploadableRequest;
 
 import static org.apache.cassandra.sidecar.common.http.SidecarHttpHeaderNames.AUTH_ROLE;
+import static org.apache.cassandra.sidecar.common.http.SidecarQueryParamNames.INSTANCE_ID;
 import static org.apache.cassandra.sidecar.common.utils.StringUtils.isNullOrEmpty;
 
 /**
@@ -252,6 +253,13 @@ protected HttpRequest<Buffer> vertxRequest(SidecarInstance sidecarInstance, Requ
                                                              sidecarInstance.hostname(),
                                                              request.requestURI());
 
+        if (config.instanceId() != null)
+        {
+            vertxRequest = vertxRequest.addQueryParam(INSTANCE_ID, String.valueOf(config.instanceId()));
+            LOGGER.debug("Appended {}={} to request uri. originalUri={}, finalUri={}",
+                         INSTANCE_ID, config.instanceId(), request.requestURI(), vertxRequest.uri());
+        }
+
         vertxRequest = applyHeaders(vertxRequest, request.headers());
 
         Map<String, String> customHeaders = context.customHeaders();

analytics-sidecar-vertx-client/src/test/java/org/apache/cassandra/sidecar/client/VertxHttpClientTest.java

@@ -22,11 +22,15 @@
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 
+import io.netty.handler.codec.http.HttpMethod;
 import io.vertx.core.Vertx;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.ext.web.client.HttpRequest;
 
+import org.apache.cassandra.sidecar.common.request.Request;
+
 import static org.apache.cassandra.sidecar.common.http.SidecarHttpHeaderNames.AUTH_ROLE;
+import static org.apache.cassandra.sidecar.common.http.SidecarQueryParamNames.INSTANCE_ID;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
@@ -56,16 +60,61 @@ public void testAuthHeaderSet()
         HttpClientConfig config = httpClientConfigBuilder().cassandraRole("custom_role").build();
         try (VertxHttpClient client = new VertxHttpClient(vertx, config))
         {
-            SidecarInstance instance = mock(SidecarInstance.class);
-            when(instance.port()).thenReturn(9043);
-            when(instance.hostname()).thenReturn("localhost");
             RequestContext context = new RequestContext.Builder().ringRequest().build();
-            HttpRequest<Buffer> request = client.vertxRequest(instance, context);
+            HttpRequest<Buffer> request = client.vertxRequest(mockInstance(), context);
             assertThat(request.headers()).isNotEmpty();
             assertThat(request.headers().get(AUTH_ROLE)).isEqualTo("custom_role");
         }
     }
 
+    @Test
+    public void testInstanceIdQueryParamAppended()
+    {
+        HttpClientConfig config = httpClientConfigBuilder().instanceId(42).build();
+        try (VertxHttpClient client = new VertxHttpClient(vertx, config))
+        {
+            RequestContext context = new RequestContext.Builder().ringRequest().build();
+            HttpRequest<Buffer> request = client.vertxRequest(mockInstance(), context);
+            assertThat(request.queryParams().get(INSTANCE_ID)).isEqualTo("42");
+        }
+    }
+
+    @Test
+    public void testInstanceIdQueryParamNotAppendedWhenNull()
+    {
+        HttpClientConfig config = httpClientConfigBuilder().build();
+        try (VertxHttpClient client = new VertxHttpClient(vertx, config))
+        {
+            RequestContext context = new RequestContext.Builder().ringRequest().build();
+            HttpRequest<Buffer> request = client.vertxRequest(mockInstance(), context);
+            assertThat(request.queryParams().contains(INSTANCE_ID)).isFalse();
+        }
+    }
+
+    @Test
+    public void testInstanceIdQueryParamAppendedWithExistingQueryParams()
+    {
+        HttpClientConfig config = httpClientConfigBuilder().instanceId(7).build();
+        try (VertxHttpClient client = new VertxHttpClient(vertx, config))
+        {
+            Request mockRequest = mock(Request.class);
+            when(mockRequest.method()).thenReturn(HttpMethod.GET);
+            when(mockRequest.requestURI()).thenReturn("/api/v1/ring?existingParam=value");
+            RequestContext context = new RequestContext.Builder().request(mockRequest).build();
+            HttpRequest<Buffer> request = client.vertxRequest(mockInstance(), context);
+            assertThat(request.queryParams().get("existingParam")).isEqualTo("value");
+            assertThat(request.queryParams().get(INSTANCE_ID)).isEqualTo("7");
+        }
+    }
+
+    private SidecarInstance mockInstance()
+    {
+        SidecarInstance instance = mock(SidecarInstance.class);
+        when(instance.port()).thenReturn(9043);
+        when(instance.hostname()).thenReturn("localhost");
+        return instance;
+    }
+
     private HttpClientConfig.Builder<?> httpClientConfigBuilder()
     {
         return new HttpClientConfig.Builder<>()

cassandra-analytics-core/src/main/java/org/apache/cassandra/clients/AnalyticsSidecarClient.java

@@ -19,6 +19,9 @@
 
 package org.apache.cassandra.clients;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import o.a.c.sidecar.client.shaded.io.vertx.core.Vertx;
 import o.a.c.sidecar.client.shaded.io.vertx.core.VertxOptions;
 import o.a.c.sidecar.client.shaded.client.HttpClientConfig;
@@ -36,6 +39,8 @@
 
 public class AnalyticsSidecarClient
 {
+    private static final Logger LOGGER = LoggerFactory.getLogger(AnalyticsSidecarClient.class);
+
     private AnalyticsSidecarClient()
     {
     }
@@ -45,20 +50,12 @@ public static SidecarClient from(SidecarInstancesProvider sidecarInstancesProvid
         Vertx vertx = Vertx.vertx(new VertxOptions().setUseDaemonThread(true)
                                                     .setWorkerPoolSize(conf.getMaxHttpConnections()));
 
-        String userAgent = transportModeBasedWriterUserAgent(conf.getTransportInfo().getTransport());
-        HttpClientConfig httpClientConfig = new HttpClientConfig.Builder<>()
-                                            .timeoutMillis(conf.getHttpResponseTimeoutMs())
-                                            .idleTimeoutMillis(conf.getHttpConnectionTimeoutMs())
-                                            .userAgent(userAgent)
-                                            .keyStoreInputStream(conf.getKeyStore())
-                                            .keyStorePassword(conf.getKeyStorePassword())
-                                            .keyStoreType(conf.getKeyStoreTypeOrDefault())
-                                            .trustStoreInputStream(conf.getTrustStore())
-                                            .trustStorePassword(conf.getTrustStorePasswordOrDefault())
-                                            .trustStoreType(conf.getTrustStoreTypeOrDefault())
-                                            .ssl(conf.hasKeystoreAndKeystorePassword())
-                                            .cassandraRole(conf.getCassandraRole())
-                                            .build();
+        HttpClientConfig httpClientConfig = buildHttpClientConfig(conf);
+        if (httpClientConfig.instanceId() != null)
+        {
+            LOGGER.info("Sidecar HTTP client configured with instanceId={} (applied to every outbound sidecar request)",
+                        httpClientConfig.instanceId());
+        }
 
         StartupValidator.instance().register(new SslValidation(conf));
         StartupValidator.instance().register(new BulkWriterKeyStoreValidation(conf));
@@ -74,6 +71,25 @@ public static SidecarClient from(SidecarInstancesProvider sidecarInstancesProvid
         return Sidecar.buildClient(sidecarConfig, vertx, httpClientConfig, sidecarInstancesProvider);
     }
 
+    static HttpClientConfig buildHttpClientConfig(BulkSparkConf conf)
+    {
+        String userAgent = transportModeBasedWriterUserAgent(conf.getTransportInfo().getTransport());
+        return new HttpClientConfig.Builder<>()
+               .timeoutMillis(conf.getHttpResponseTimeoutMs())
+               .idleTimeoutMillis(conf.getHttpConnectionTimeoutMs())
+               .userAgent(userAgent)
+               .keyStoreInputStream(conf.getKeyStore())
+               .keyStorePassword(conf.getKeyStorePassword())
+               .keyStoreType(conf.getKeyStoreTypeOrDefault())
+               .trustStoreInputStream(conf.getTrustStore())
+               .trustStorePassword(conf.getTrustStorePasswordOrDefault())
+               .trustStoreType(conf.getTrustStoreTypeOrDefault())
+               .ssl(conf.hasKeystoreAndKeystorePassword())
+               .cassandraRole(conf.getCassandraRole())
+               .instanceId(conf.getSidecarInstanceId())
+               .build();
+    }
+
     static String transportModeBasedWriterUserAgent(DataTransport transport)
     {
         switch (transport)

cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConf.java

@@ -125,6 +125,7 @@ public class BulkSparkConf implements Serializable
     public static final String SIDECAR_REQUEST_RETRY_DELAY_MILLIS      = SETTING_PREFIX + "sidecar.request.retries.delay.milliseconds";
     public static final String SIDECAR_REQUEST_MAX_RETRY_DELAY_MILLIS  = SETTING_PREFIX + "sidecar.request.retries.max.delay.milliseconds";
     public static final String SIDECAR_REQUEST_TIMEOUT_SECONDS         = SETTING_PREFIX + "sidecar.request.timeout.seconds";
+    public static final String SIDECAR_INSTANCE_ID                     = SETTING_PREFIX + "sidecar.instance.id";
     public static final String SKIP_CLEAN                              = SETTING_PREFIX + "job.skip_clean";
     public static final String USE_OPENSSL                             = SETTING_PREFIX + "use_openssl";
     // defines the max number of consecutive retries allowed in the ring monitor
@@ -579,6 +580,21 @@ public int getSidecarRequestTimeoutSeconds()
         return getInt(SIDECAR_REQUEST_TIMEOUT_SECONDS, DEFAULT_SIDECAR_REQUEST_TIMEOUT_SECONDS);
     }
 
+    @Nullable
+    public Integer getSidecarInstanceId()
+    {
+        Integer value = getOptionalInt(SIDECAR_INSTANCE_ID).orElse(null);
+        // Validated here (not just in HttpClientConfig.Builder.instanceId()) to surface a
+        // Spark-conf-specific error message before any HTTP client is constructed. Keep this
+        // constraint (>= 0) in sync with that check.
+        if (value != null && value < 0)
+        {
+            throw new IllegalArgumentException("Spark conf " + SIDECAR_INSTANCE_ID
+                                               + " must be a non-negative integer; got " + value);
+        }
+        return value;
+    }
+
     public int getHttpConnectionTimeoutMs()
     {
         return getInt(HTTP_CONNECTION_TIMEOUT, DEFAULT_HTTP_CONNECTION_TIMEOUT);