diff --git a/CHANGES.txt b/CHANGES.txt
index c3cda0a5a..32cb3f0ff 100644
--- a/CHANGES.txt
+++ b/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)
* CDC reader stats silently dropped in SidecarCdcBuilder (CASSANALYTICS-191)
* Add CapturePublishedSchema metric to SidecarCdcStats (CASSANALYTICS-189)
* Expand list of architecture that supports unaligned access in FastByteOperations (CASSANALYTICS-188)
diff --git a/analytics-sidecar-client-common/src/main/java/org/apache/cassandra/sidecar/client/SidecarInstance.java b/analytics-sidecar-client-common/src/main/java/org/apache/cassandra/sidecar/client/SidecarInstance.java
index 1d2d14829..26b977b02 100644
--- a/analytics-sidecar-client-common/src/main/java/org/apache/cassandra/sidecar/client/SidecarInstance.java
+++ b/analytics-sidecar-client-common/src/main/java/org/apache/cassandra/sidecar/client/SidecarInstance.java
@@ -32,4 +32,20 @@ public interface SidecarInstance
* @return the hostname where the Cassandra Sidecar instance is running
*/
String hostname();
+
+ /**
+ * Returns the identifier of the specific Cassandra instance that requests sent to this Sidecar
+ * endpoint should be routed to, or {@code null} when no per-instance identifier is configured.
+ *
+ * When non-null, this value is used to populate the {@code instanceId} query parameter on outbound
+ * requests so the Sidecar can resolve the correct local Cassandra instance even when a shared address
+ * (for example a load balancer) hides the real target from the {@code Host} header. When {@code null},
+ * the client falls back to the job-level {@code instanceId} configured on the HTTP client, if any.
+ *
+ * @return the per-instance identifier, or {@code null} when not set
+ */
+ default Integer instanceId()
+ {
+ return null;
+ }
}
diff --git a/analytics-sidecar-client-common/src/main/java/org/apache/cassandra/sidecar/client/SidecarInstanceImpl.java b/analytics-sidecar-client-common/src/main/java/org/apache/cassandra/sidecar/client/SidecarInstanceImpl.java
index 751218b16..f3bbf3b1b 100644
--- a/analytics-sidecar-client-common/src/main/java/org/apache/cassandra/sidecar/client/SidecarInstanceImpl.java
+++ b/analytics-sidecar-client-common/src/main/java/org/apache/cassandra/sidecar/client/SidecarInstanceImpl.java
@@ -27,22 +27,45 @@ public class SidecarInstanceImpl implements SidecarInstance
{
protected int port;
protected String hostname;
+ protected Integer instanceId;
/**
- * Constructs a new Sidecar instance with the given {@code port} and {@code hostname}
+ * Constructs a new Sidecar instance with the given {@code port} and {@code hostname} and no
+ * per-instance identifier (requests fall back to the job-level {@code instanceId}, if any).
*
* @param hostname the host name where Sidecar is running
* @param port the port where Sidecar is running
*/
public SidecarInstanceImpl(String hostname, int port)
+ {
+ this(hostname, port, null);
+ }
+
+ /**
+ * Constructs a new Sidecar instance with the given {@code hostname}, {@code port} and per-instance
+ * {@code instanceId}.
+ *
+ * @param hostname the host name where Sidecar is running
+ * @param port the port where Sidecar is running
+ * @param instanceId the identifier of the Cassandra instance that requests sent to this Sidecar
+ * endpoint should be routed to, or {@code null} to fall back to the job-level
+ * {@code instanceId}
+ */
+ public SidecarInstanceImpl(String hostname, int port, Integer instanceId)
{
if (port < 1 || port > 65535)
{
throw new IllegalArgumentException(String.format("Invalid port number for the Sidecar service: %d",
port));
}
+ if (instanceId != null && instanceId < 0)
+ {
+ throw new IllegalArgumentException(String.format("Invalid instanceId for the Sidecar service: %d",
+ instanceId));
+ }
this.port = port;
this.hostname = Objects.requireNonNull(hostname, "The Sidecar hostname must be non-null");
+ this.instanceId = instanceId;
}
/**
@@ -63,6 +86,15 @@ public String hostname()
return hostname;
}
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Integer instanceId()
+ {
+ return instanceId;
+ }
+
/**
* {@inheritDoc}
*/
@@ -78,7 +110,7 @@ public boolean equals(Object o)
return false;
}
SidecarInstanceImpl that = (SidecarInstanceImpl) o;
- return port == that.port && Objects.equals(hostname, that.hostname);
+ return port == that.port && Objects.equals(hostname, that.hostname) && Objects.equals(instanceId, that.instanceId);
}
/**
@@ -87,7 +119,7 @@ public boolean equals(Object o)
@Override
public int hashCode()
{
- return Objects.hash(port, hostname);
+ return Objects.hash(port, hostname, instanceId);
}
/**
@@ -99,6 +131,7 @@ public String toString()
return "SidecarInstanceImpl{" +
"port=" + port +
", hostname='" + hostname + '\'' +
+ ", instanceId=" + instanceId +
'}';
}
}
diff --git a/analytics-sidecar-client-common/src/main/java/org/apache/cassandra/sidecar/common/http/SidecarQueryParamNames.java b/analytics-sidecar-client-common/src/main/java/org/apache/cassandra/sidecar/common/http/SidecarQueryParamNames.java
new file mode 100644
index 000000000..6b615d261
--- /dev/null
+++ b/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}).
+ *
+ *
Requires a Sidecar server >= 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()
+ {
+ }
+}
diff --git a/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/HttpClientConfig.java b/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/HttpClientConfig.java
index 5d2714b3c..ae2042367 100644
--- a/analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/HttpClientConfig.java
+++ b/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>
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.
*
diff --git a/analytics-sidecar-client/src/test/java/org/apache/cassandra/sidecar/client/HttpClientConfigTest.java b/analytics-sidecar-client/src/test/java/org/apache/cassandra/sidecar/client/HttpClientConfigTest.java
index a2c1db659..0a839a65d 100644
--- a/analytics-sidecar-client/src/test/java/org/apache/cassandra/sidecar/client/HttpClientConfigTest.java
+++ b/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");
+ }
}
diff --git a/analytics-sidecar-client/src/test/java/org/apache/cassandra/sidecar/client/SidecarInstanceImplTest.java b/analytics-sidecar-client/src/test/java/org/apache/cassandra/sidecar/client/SidecarInstanceImplTest.java
index 8307eba75..eac97872e 100644
--- a/analytics-sidecar-client/src/test/java/org/apache/cassandra/sidecar/client/SidecarInstanceImplTest.java
+++ b/analytics-sidecar-client/src/test/java/org/apache/cassandra/sidecar/client/SidecarInstanceImplTest.java
@@ -18,6 +18,11 @@
package org.apache.cassandra.sidecar.client;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+
/**
* Unit tests for the {@link SidecarInstanceImpl} class
*/
@@ -28,4 +33,39 @@ protected SidecarInstance newInstance(String hostname, int port)
{
return new SidecarInstanceImpl(hostname, port);
}
+
+ @Test
+ void testInstanceIdDefaultsToNull()
+ {
+ assertThat(new SidecarInstanceImpl("localhost", 8080).instanceId()).isNull();
+ }
+
+ @Test
+ void testInstanceIdIsRetained()
+ {
+ assertThat(new SidecarInstanceImpl("localhost", 8080, 2).instanceId()).isEqualTo(2);
+ assertThat(new SidecarInstanceImpl("localhost", 8080, 0).instanceId()).isEqualTo(0);
+ }
+
+ @Test
+ void testNegativeInstanceIdRejected()
+ {
+ assertThatExceptionOfType(IllegalArgumentException.class)
+ .isThrownBy(() -> new SidecarInstanceImpl("localhost", 8080, -1))
+ .withMessageContaining("Invalid instanceId for the Sidecar service: -1");
+ }
+
+ @Test
+ void testEqualityDistinguishesInstanceId()
+ {
+ SidecarInstance a = new SidecarInstanceImpl("localhost", 8080, 1);
+ SidecarInstance b = new SidecarInstanceImpl("localhost", 8080, 2);
+ SidecarInstance c = new SidecarInstanceImpl("localhost", 8080, 1);
+ SidecarInstance noId = new SidecarInstanceImpl("localhost", 8080);
+
+ assertThat(a).isEqualTo(c);
+ assertThat(a).hasSameHashCodeAs(c);
+ assertThat(a).isNotEqualTo(b);
+ assertThat(a).isNotEqualTo(noId);
+ }
}
diff --git a/analytics-sidecar-vertx-client/src/main/java/org/apache/cassandra/sidecar/client/VertxHttpClient.java b/analytics-sidecar-vertx-client/src/main/java/org/apache/cassandra/sidecar/client/VertxHttpClient.java
index f4c3890df..7bf7842a5 100644
--- a/analytics-sidecar-vertx-client/src/main/java/org/apache/cassandra/sidecar/client/VertxHttpClient.java
+++ b/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,18 @@ protected HttpRequest vertxRequest(SidecarInstance sidecarInstance, Requ
sidecarInstance.hostname(),
request.requestURI());
+ // Prefer the id carried by the specific instance this request is being sent to, so requests
+ // fanned out across multiple instances each get the correct id. Fall back to the job-level
+ // id from the HTTP client config only when the instance does not carry its own.
+ Integer instanceId = sidecarInstance.instanceId() != null ? sidecarInstance.instanceId() : config.instanceId();
+ if (instanceId != null)
+ {
+ vertxRequest = vertxRequest.addQueryParam(INSTANCE_ID, String.valueOf(instanceId));
+ LOGGER.debug("Appended {}={} to request uri. instance={}:{}, originalUri={}, finalUri={}",
+ INSTANCE_ID, instanceId, sidecarInstance.hostname(), sidecarInstance.port(),
+ request.requestURI(), vertxRequest.uri());
+ }
+
vertxRequest = applyHeaders(vertxRequest, request.headers());
Map customHeaders = context.customHeaders();
diff --git a/analytics-sidecar-vertx-client/src/test/java/org/apache/cassandra/sidecar/client/VertxHttpClientTest.java b/analytics-sidecar-vertx-client/src/test/java/org/apache/cassandra/sidecar/client/VertxHttpClientTest.java
index 8d8d0952d..5c807fba2 100644
--- a/analytics-sidecar-vertx-client/src/test/java/org/apache/cassandra/sidecar/client/VertxHttpClientTest.java
+++ b/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,92 @@ 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 request = client.vertxRequest(instance, context);
+ HttpRequest 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 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 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 request = client.vertxRequest(mockInstance(), context);
+ assertThat(request.queryParams().get("existingParam")).isEqualTo("value");
+ assertThat(request.queryParams().get(INSTANCE_ID)).isEqualTo("7");
+ }
+ }
+
+ @Test
+ public void testPerInstanceIdOverridesGlobalInstanceId()
+ {
+ HttpClientConfig config = httpClientConfigBuilder().instanceId(1).build();
+ try (VertxHttpClient client = new VertxHttpClient(vertx, config))
+ {
+ RequestContext context = new RequestContext.Builder().ringRequest().build();
+ // The instance carries its own id (3), which must win over the job-level id (1).
+ HttpRequest request = client.vertxRequest(mockInstance(3), context);
+ assertThat(request.queryParams().get(INSTANCE_ID)).isEqualTo("3");
+ }
+ }
+
+ @Test
+ public void testPerInstanceIdUsedWhenGlobalInstanceIdIsNull()
+ {
+ HttpClientConfig config = httpClientConfigBuilder().build();
+ try (VertxHttpClient client = new VertxHttpClient(vertx, config))
+ {
+ RequestContext context = new RequestContext.Builder().ringRequest().build();
+ HttpRequest request = client.vertxRequest(mockInstance(5), context);
+ assertThat(request.queryParams().get(INSTANCE_ID)).isEqualTo("5");
+ }
+ }
+
+ private SidecarInstance mockInstance()
+ {
+ return mockInstance(null);
+ }
+
+ private SidecarInstance mockInstance(Integer instanceId)
+ {
+ SidecarInstance instance = mock(SidecarInstance.class);
+ when(instance.port()).thenReturn(9043);
+ when(instance.hostname()).thenReturn("localhost");
+ when(instance.instanceId()).thenReturn(instanceId);
+ return instance;
+ }
+
private HttpClientConfig.Builder httpClientConfigBuilder()
{
return new HttpClientConfig.Builder<>()
diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/common/model/CassandraInstance.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/common/model/CassandraInstance.java
index fed2f2d59..2783d0224 100644
--- a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/common/model/CassandraInstance.java
+++ b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/common/model/CassandraInstance.java
@@ -55,4 +55,20 @@ public interface CassandraInstance extends TokenOwner
* @return status of the node
*/
NodeStatus nodeStatus();
+
+ /**
+ * Returns the identifier of the specific Cassandra instance that a shared Sidecar endpoint should route
+ * requests to, or {@code null} when not configured.
+ *
+ * This is only meaningful when a single Sidecar endpoint (for example, one fronted by a load balancer)
+ * fronts more than one Cassandra instance: the id disambiguates which local instance a request targets,
+ * since the endpoint alone (hostname/Host header) cannot. When {@code null}, callers fall back to a
+ * job-level default, if any.
+ *
+ * @return the per-instance Sidecar routing id, or {@code null} when not set
+ */
+ default Integer sidecarInstanceId()
+ {
+ return null;
+ }
}
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/clients/AnalyticsSidecarClient.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/clients/AnalyticsSidecarClient.java
index 2096a60a9..810739633 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/clients/AnalyticsSidecarClient.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/clients/AnalyticsSidecarClient.java
@@ -19,12 +19,18 @@
package org.apache.cassandra.clients;
+import java.util.List;
+
+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;
import o.a.c.sidecar.client.shaded.client.SidecarClient;
import o.a.c.sidecar.client.shaded.client.SidecarClientConfig;
import o.a.c.sidecar.client.shaded.client.SidecarClientConfigImpl;
+import o.a.c.sidecar.client.shaded.client.SidecarInstance;
import o.a.c.sidecar.client.shaded.client.SidecarInstancesProvider;
import org.apache.cassandra.spark.bulkwriter.BulkSparkConf;
import org.apache.cassandra.spark.bulkwriter.DataTransport;
@@ -36,6 +42,8 @@
public class AnalyticsSidecarClient
{
+ private static final Logger LOGGER = LoggerFactory.getLogger(AnalyticsSidecarClient.class);
+
private AnalyticsSidecarClient()
{
}
@@ -45,20 +53,8 @@ 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);
+ warnIfGlobalInstanceIdIsAmbiguous(httpClientConfig, sidecarInstancesProvider);
StartupValidator.instance().register(new SslValidation(conf));
StartupValidator.instance().register(new BulkWriterKeyStoreValidation(conf));
@@ -74,6 +70,60 @@ public static SidecarClient from(SidecarInstancesProvider sidecarInstancesProvid
return Sidecar.buildClient(sidecarConfig, vertx, httpClientConfig, sidecarInstancesProvider);
}
+ /**
+ * Warns when a single job-level {@code instanceId} would be stamped uniformly onto requests fanned out
+ * across more than one sidecar instance, none of which carry their own per-instance id. That is only correct
+ * when every instance resolves the same id (for example a 1:1 Cassandra-to-Sidecar deployment where each local
+ * instance is id {@code 1}); otherwise requests are misrouted. Operators should instead assign a per-instance id
+ * to each sidecar contact point (see {@link org.apache.cassandra.spark.common.SidecarInstanceFactory}).
+ */
+ static void warnIfGlobalInstanceIdIsAmbiguous(HttpClientConfig httpClientConfig,
+ SidecarInstancesProvider sidecarInstancesProvider)
+ {
+ Integer globalInstanceId = httpClientConfig.instanceId();
+ if (globalInstanceId == null)
+ {
+ return;
+ }
+
+ List instances = sidecarInstancesProvider.instances();
+ boolean anyPerInstanceId = instances.stream().anyMatch(instance -> instance.instanceId() != null);
+ if (instances.size() > 1 && !anyPerInstanceId)
+ {
+ LOGGER.warn("Spark conf {}={} will be applied uniformly to every request across {} sidecar instances, "
+ + "none of which declare their own instanceId. This is only correct when every instance "
+ + "resolves the same id (for example a 1:1 Cassandra-to-Sidecar deployment where each local "
+ + "instance is id {}). If the instances have distinct ids this misroutes requests; assign a "
+ + "per-instance id to each sidecar contact point (host:port={}) instead.",
+ BulkSparkConf.SIDECAR_INSTANCE_ID, globalInstanceId, instances.size(),
+ globalInstanceId, globalInstanceId);
+ }
+ else
+ {
+ LOGGER.info("Sidecar HTTP client configured with job-level instanceId={} (used only for requests to "
+ + "instances without their own instanceId)", globalInstanceId);
+ }
+ }
+
+ 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)
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConf.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConf.java
index 6e64f145a..cddcc1ebc 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConf.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConf.java
@@ -126,6 +126,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
@@ -582,6 +583,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);
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java
index 3f7523bc0..67d7eb3bc 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfo.java
@@ -253,7 +253,8 @@ void validateTimeSkewWithLocalNow(Range range, Instant localNow) thr
.stream()
.flatMap(Collection::stream)
.distinct() // remove duplications
- .map(replica -> new SidecarInstanceImpl(replica.nodeName(), getCassandraContext().sidecarPort()))
+ .map(replica -> new SidecarInstanceImpl(replica.nodeName(), getCassandraContext().sidecarPort(),
+ replica.sidecarInstanceId()))
.collect(Collectors.toList());
timeSkew = getCassandraContext().getSidecarClient().timeSkew(instances).get();
}
@@ -484,9 +485,70 @@ protected WriteAvailability determineWriteAvailability(RingInstance instance)
private TokenRangeMapping getTokenRangeReplicasFromSidecar()
{
- return TokenRangeMapping.create(this::getTokenRangesAndReplicaSets,
- this::getPartitioner,
- metadata -> new RingInstance(metadata, clusterId()));
+ // Resolve per-instance ids from the contact points actually in effect for this cluster (getCluster()
+ // already picks the right source: conf.sidecarContactPoints() for a plain job, or
+ // conf.coordinatedWriteConf().cluster(clusterId).sidecarContactPoints() for a coordinated-write job).
+ // Deriving this from conf.sidecarContactPoints() directly would silently miss (or NPE on) coordinated
+ // writes, since their contact points don't live there.
+ Map instanceIdsByHostname = sidecarInstanceIdsByHostname(getCassandraContext().getCluster());
+ TokenRangeMapping topology =
+ TokenRangeMapping.create(this::getTokenRangesAndReplicaSets,
+ this::getPartitioner,
+ metadata -> new RingInstance(metadata, clusterId(),
+ instanceIdsByHostname.get(metadata.fqdn())));
+ validateSidecarInstanceIdCoverage(topology.allInstances());
+ return topology;
+ }
+
+ /**
+ * Builds a hostname (nodeName/fqdn) to per-instance Sidecar routing id lookup from the given contact points,
+ * e.g. ones declared as {@code "host:port="} (see {@link SidecarInstanceFactory#createFromString}).
+ *
+ * @param contactPoints the Sidecar contact points in effect for this cluster
+ * @return a map of hostname to configured Sidecar instance id; entries with no configured id are omitted
+ */
+ @VisibleForTesting
+ static Map sidecarInstanceIdsByHostname(Set contactPoints)
+ {
+ return contactPoints.stream()
+ .filter(instance -> instance.instanceId() != null)
+ .collect(Collectors.toMap(SidecarInstance::hostname, SidecarInstance::instanceId));
+ }
+
+ /**
+ * Guards against the data-correctness risk of a single job-level {@code instanceId} (see
+ * {@link BulkSparkConf#SIDECAR_INSTANCE_ID}) being stamped uniformly onto requests fanned out across more than
+ * one real Cassandra instance. That is only correct when every instance in the ring resolves its own id (see
+ * {@link #sidecarInstanceIdsByHostname}); otherwise requests to the unresolved instances would be silently
+ * misrouted to whichever instance the global id happens to identify.
+ *
+ * @param instances the distinct instances discovered from the live ring
+ */
+ @VisibleForTesting
+ void validateSidecarInstanceIdCoverage(Set instances)
+ {
+ Integer globalInstanceId = conf.getSidecarInstanceId();
+ if (globalInstanceId == null || instances.size() <= 1)
+ {
+ return;
+ }
+
+ List unresolvedInstances = instances.stream()
+ .filter(instance -> instance.sidecarInstanceId() == null)
+ .map(RingInstance::nodeName)
+ .sorted()
+ .collect(Collectors.toList());
+ if (!unresolvedInstances.isEmpty())
+ {
+ throw new IllegalStateException(
+ String.format("Ambiguous Sidecar instanceId configuration: Spark conf %s=%d would be applied uniformly "
+ + "to %d/%d ring instances that have no per-instance id configured (%s). This misroutes "
+ + "requests whenever a single Sidecar endpoint fronts more than one of these instances "
+ + "(for example, behind a load balancer). Configure a per-instance id for each affected "
+ + "instance instead, using the host[:port]= syntax in %s.",
+ BulkSparkConf.SIDECAR_INSTANCE_ID, globalInstanceId, unresolvedInstances.size(), instances.size(),
+ unresolvedInstances, WriterOptions.SIDECAR_CONTACT_POINTS.name()));
+ }
}
public String getVersionFromFeature()
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/RingInstance.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/RingInstance.java
index 5e5ed60e9..03fe005fd 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/RingInstance.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/RingInstance.java
@@ -39,10 +39,22 @@ public class RingInstance implements CassandraInstance, Serializable
private static final long serialVersionUID = 4399143234683369652L;
private RingEntry ringEntry;
private @Nullable String clusterId;
+ private @Nullable Integer sidecarInstanceId;
public RingInstance(ReplicaMetadata replica, @Nullable String clusterId)
+ {
+ this(replica, clusterId, null);
+ }
+
+ /**
+ * @param sidecarInstanceId the id of the Cassandra instance that a shared Sidecar endpoint fronting this
+ * instance should route requests to, or {@code null} when not configured for this
+ * instance (see {@link CassandraInstance#sidecarInstanceId()})
+ */
+ public RingInstance(ReplicaMetadata replica, @Nullable String clusterId, @Nullable Integer sidecarInstanceId)
{
this.clusterId = clusterId;
+ this.sidecarInstanceId = sidecarInstanceId;
this.ringEntry = new RingEntry.Builder()
.fqdn(replica.fqdn())
.address(replica.address())
@@ -61,8 +73,15 @@ public RingInstance(RingEntry ringEntry)
@VisibleForTesting
public RingInstance(RingEntry ringEntry, @Nullable String clusterId)
+ {
+ this(ringEntry, clusterId, null);
+ }
+
+ @VisibleForTesting
+ public RingInstance(RingEntry ringEntry, @Nullable String clusterId, @Nullable Integer sidecarInstanceId)
{
this.clusterId = clusterId;
+ this.sidecarInstanceId = sidecarInstanceId;
this.ringEntry = ringEntry;
}
@@ -122,13 +141,21 @@ public NodeStatus nodeStatus()
return NodeStatus.fromNameIgnoreCase(ringEntry.status());
}
+ @Override
+ @Nullable
+ public Integer sidecarInstanceId()
+ {
+ return sidecarInstanceId;
+ }
+
/**
* Custom equality that compares the token, fully qualified domain name, the rack, the port, the datacenter
* and the clusterId
*
- * Note that node state, status and IP address are not part of the calculation. The IP address is excluded
- * because a node can come back with a different IP address (e.g. a pod replacement in Kubernetes) while
- * remaining the same logical instance.
+ * Note that node state, status, IP address and sidecarInstanceId are not part of the calculation. The IP
+ * address is excluded because a node can come back with a different IP address (e.g. a pod replacement in
+ * Kubernetes) while remaining the same logical instance. sidecarInstanceId is excluded because it is routing
+ * metadata derived from configuration, not part of the instance's identity.
*
* @param other the other instance
* @return true if both instances are equal, false otherwise
@@ -171,7 +198,7 @@ public int hashCode()
@Override
public String toString()
{
- return "RingInstance{cluster='" + clusterId + "', " + ringEntry.toString() + '}';
+ return "RingInstance{cluster='" + clusterId + "', sidecarInstanceId=" + sidecarInstanceId + ", " + ringEntry.toString() + '}';
}
public RingEntry ringEntry()
@@ -194,6 +221,7 @@ private void writeObject(ObjectOutputStream out) throws IOException
out.writeObject(ringEntry.load());
out.writeObject(ringEntry.owns());
out.writeObject(clusterId);
+ out.writeObject(sidecarInstanceId);
}
private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException
@@ -211,6 +239,7 @@ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundE
String load = (String) in.readObject();
String owns = (String) in.readObject();
String clusterId = (String) in.readObject();
+ Integer sidecarInstanceId = (Integer) in.readObject();
ringEntry = new RingEntry.Builder().datacenter(datacenter)
.address(address)
.port(port)
@@ -224,5 +253,6 @@ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundE
.owns(owns)
.build();
this.clusterId = clusterId;
+ this.sidecarInstanceId = sidecarInstanceId;
}
}
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/SidecarDataTransferApi.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/SidecarDataTransferApi.java
index 9199b01ca..2d98c5d27 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/SidecarDataTransferApi.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/SidecarDataTransferApi.java
@@ -161,6 +161,6 @@ protected String getUploadId(String sessionID, String jobId)
protected SidecarInstanceImpl toSidecarInstance(CassandraInstance instance)
{
- return new SidecarInstanceImpl(instance.nodeName(), sidecarPort);
+ return new SidecarInstanceImpl(instance.nodeName(), sidecarPort, instance.sidecarInstanceId());
}
}
diff --git a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/common/SidecarInstanceFactory.java b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/common/SidecarInstanceFactory.java
index 129826b50..798121a88 100644
--- a/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/common/SidecarInstanceFactory.java
+++ b/cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/common/SidecarInstanceFactory.java
@@ -38,6 +38,11 @@ private SidecarInstanceFactory()
/**
* Create SidecarInstance object by parsing the input string, which is IP address or hostname and optionally includes port
+ * The input may also carry an optional per-instance id as a trailing {@code "="} suffix, e.g.
+ * {@code "host:9043=2"}. The id identifies which local Cassandra instance the receiving Sidecar should route
+ * requests to; it is used to populate the {@code instanceId} query parameter per instance instead of relying on a
+ * single job-level value. {@code '='} cannot appear in a hostname, IPv4/IPv6 address or port, so the suffix is
+ * unambiguous. When absent, requests fall back to the job-level {@code instanceId}, if any.
* @param input hostname string that can optionally includes the port. If port is present, the defaultPort param is ignored.
* @param defaultPort port value used when the input string contains no port
* @return SidecarInstanceImpl
@@ -46,22 +51,42 @@ public static SidecarInstanceImpl createFromString(String input, int defaultPort
{
Preconditions.checkArgument(StringUtils.isNotEmpty(input), "Unable to create sidecar instance from empty input");
- String hostname = input;
+ String address = input;
+ Integer instanceId = null;
+ // Optional per-instance id, expressed as a trailing "=" suffix (e.g. "host:9043=2").
+ int equalsIndex = input.lastIndexOf('=');
+ if (equalsIndex >= 0)
+ {
+ String instanceIdStr = input.substring(equalsIndex + 1).trim();
+ try
+ {
+ instanceId = Integer.parseInt(instanceIdStr);
+ }
+ catch (NumberFormatException e)
+ {
+ throw new IllegalArgumentException(
+ String.format("Invalid sidecar instanceId '%s' in '%s'; expected a non-negative integer", instanceIdStr, input), e);
+ }
+ Preconditions.checkArgument(instanceId >= 0, "Sidecar instanceId must be non-negative; got %s in '%s'", instanceId, input);
+ address = input.substring(0, equalsIndex);
+ }
+
+ String hostname = address;
int port = defaultPort;
// has port in the string. The former matches ipv6 and the latter matches ipv4 and hostnames
// ipv6 with port example: [2024:a::1]:8080
- if (input.contains("]:") || (!input.startsWith("[") && input.contains(":")))
+ if (address.contains("]:") || (!address.startsWith("[") && address.contains(":")))
{
- int index = input.lastIndexOf(':');
- hostname = input.substring(0, index); // includes ']' if it is ipv6
- String portStr = input.substring(index + 1);
+ int index = address.lastIndexOf(':');
+ hostname = address.substring(0, index); // includes ']' if it is ipv6
+ String portStr = address.substring(index + 1);
port = Integer.parseInt(portStr);
}
Preconditions.checkState(port != -1, "Unable to resolve port from %s", input);
- LOGGER.info("Create sidecar instance. hostname={} port={}", hostname, port);
- return new SidecarInstanceImpl(hostname, port);
+ LOGGER.info("Create sidecar instance. hostname={} port={} instanceId={}", hostname, port, instanceId);
+ return new SidecarInstanceImpl(hostname, port, instanceId);
}
/**
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/clients/AnalyticsSidecarClientTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/clients/AnalyticsSidecarClientTest.java
new file mode 100644
index 000000000..e70bc0d08
--- /dev/null
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/clients/AnalyticsSidecarClientTest.java
@@ -0,0 +1,132 @@
+/*
+ * 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.clients;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Map;
+
+import com.google.common.collect.Maps;
+import org.junit.jupiter.api.Test;
+
+import o.a.c.sidecar.client.shaded.client.HttpClientConfig;
+import o.a.c.sidecar.client.shaded.client.SidecarInstanceImpl;
+import o.a.c.sidecar.client.shaded.client.SimpleSidecarInstancesProvider;
+import org.apache.cassandra.spark.bulkwriter.BulkSparkConf;
+import org.apache.cassandra.spark.bulkwriter.WriterOptions;
+import org.apache.spark.SparkConf;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+/**
+ * Unit tests for {@link AnalyticsSidecarClient}
+ */
+class AnalyticsSidecarClientTest
+{
+ @Test
+ void testBuildHttpClientConfigDefaultInstanceIdIsNull()
+ {
+ BulkSparkConf conf = new BulkSparkConf(new SparkConf(), defaultOptions());
+ HttpClientConfig httpClientConfig = AnalyticsSidecarClient.buildHttpClientConfig(conf);
+ assertThat(httpClientConfig.instanceId()).isNull();
+ }
+
+ @Test
+ void testBuildHttpClientConfigWiresInstanceId()
+ {
+ SparkConf sparkConf = new SparkConf().set(BulkSparkConf.SIDECAR_INSTANCE_ID, "9");
+ BulkSparkConf conf = new BulkSparkConf(sparkConf, defaultOptions());
+ HttpClientConfig httpClientConfig = AnalyticsSidecarClient.buildHttpClientConfig(conf);
+ assertThat(httpClientConfig.instanceId()).isEqualTo(9);
+ }
+
+ @Test
+ void testWarnIfGlobalInstanceIdIsAmbiguousAllowsSingleInstance()
+ {
+ HttpClientConfig httpClientConfig = configWithGlobalInstanceId("1");
+ SimpleSidecarInstancesProvider provider =
+ new SimpleSidecarInstancesProvider(Collections.singletonList(new SidecarInstanceImpl("127.0.0.1", 9999)));
+
+ // A single instance is unambiguous: the global id can only apply to it, so this must not fail.
+ assertThatCode(() -> AnalyticsSidecarClient.warnIfGlobalInstanceIdIsAmbiguous(httpClientConfig, provider))
+ .doesNotThrowAnyException();
+ }
+
+ @Test
+ void testWarnIfGlobalInstanceIdIsAmbiguousDoesNotThrowForMultipleInstances()
+ {
+ HttpClientConfig httpClientConfig = configWithGlobalInstanceId("2");
+ SimpleSidecarInstancesProvider provider =
+ new SimpleSidecarInstancesProvider(Arrays.asList(new SidecarInstanceImpl("127.0.0.1", 9999),
+ new SidecarInstanceImpl("127.0.0.2", 9999),
+ new SidecarInstanceImpl("127.0.0.3", 9999)));
+
+ // Multiple instances relying on a single global id is a warning (per-instance ids can override it),
+ // not a hard failure - so the job must still be allowed to start.
+ assertThatCode(() -> AnalyticsSidecarClient.warnIfGlobalInstanceIdIsAmbiguous(httpClientConfig, provider))
+ .doesNotThrowAnyException();
+ }
+
+ @Test
+ void testWarnIfGlobalInstanceIdIsAmbiguousWithPerInstanceIds()
+ {
+ HttpClientConfig httpClientConfig = configWithGlobalInstanceId("1");
+ SimpleSidecarInstancesProvider provider =
+ new SimpleSidecarInstancesProvider(Arrays.asList(new SidecarInstanceImpl("127.0.0.1", 9999, 1),
+ new SidecarInstanceImpl("127.0.0.2", 9999, 2),
+ new SidecarInstanceImpl("127.0.0.3", 9999, 3)));
+
+ assertThatCode(() -> AnalyticsSidecarClient.warnIfGlobalInstanceIdIsAmbiguous(httpClientConfig, provider))
+ .doesNotThrowAnyException();
+ }
+
+ @Test
+ void testWarnIfGlobalInstanceIdIsAmbiguousNoopWhenUnset()
+ {
+ HttpClientConfig httpClientConfig = AnalyticsSidecarClient.buildHttpClientConfig(
+ new BulkSparkConf(new SparkConf(), defaultOptions()));
+ SimpleSidecarInstancesProvider provider =
+ new SimpleSidecarInstancesProvider(Arrays.asList(new SidecarInstanceImpl("127.0.0.1", 9999),
+ new SidecarInstanceImpl("127.0.0.2", 9999)));
+
+ assertThat(httpClientConfig.instanceId()).isNull();
+ assertThatCode(() -> AnalyticsSidecarClient.warnIfGlobalInstanceIdIsAmbiguous(httpClientConfig, provider))
+ .doesNotThrowAnyException();
+ }
+
+ private HttpClientConfig configWithGlobalInstanceId(String instanceId)
+ {
+ SparkConf sparkConf = new SparkConf().set(BulkSparkConf.SIDECAR_INSTANCE_ID, instanceId);
+ return AnalyticsSidecarClient.buildHttpClientConfig(new BulkSparkConf(sparkConf, defaultOptions()));
+ }
+
+ private Map defaultOptions()
+ {
+ Map options = Maps.newTreeMap(String.CASE_INSENSITIVE_ORDER);
+ options.put(WriterOptions.SIDECAR_CONTACT_POINTS.name(), "127.0.0.1");
+ options.put(WriterOptions.KEYSPACE.name(), "ks");
+ options.put(WriterOptions.TABLE.name(), "table");
+ options.put(WriterOptions.KEYSTORE_PASSWORD.name(), "dummy_password");
+ // Base64 of "dummy"; getKeyStore() only decodes it, it never parses the bytes as a real keystore.
+ options.put(WriterOptions.KEYSTORE_BASE64_ENCODED.name(), "ZHVtbXk=");
+ return options;
+ }
+}
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConfTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConfTest.java
index 5fbca3504..a64c500cc 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConfTest.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/BulkSparkConfTest.java
@@ -169,6 +169,46 @@ void testSkipClean()
assertThat(bulkSparkConf.getSkipClean()).isTrue();
}
+ @Test
+ void testDefaultSidecarInstanceId()
+ {
+ assertThat(bulkSparkConf.getSidecarInstanceId()).isNull();
+ }
+
+ @Test
+ void testSidecarInstanceId()
+ {
+ sparkConf.set(BulkSparkConf.SIDECAR_INSTANCE_ID, "3");
+ assertThat(bulkSparkConf.getSidecarInstanceId()).isEqualTo(3);
+ }
+
+ @Test
+ void testSidecarInstanceIdZeroIsAllowed()
+ {
+ sparkConf.set(BulkSparkConf.SIDECAR_INSTANCE_ID, "0");
+ assertThat(bulkSparkConf.getSidecarInstanceId()).isEqualTo(0);
+ }
+
+ @Test
+ void testSidecarInstanceIdNegativeThrows()
+ {
+ sparkConf.set(BulkSparkConf.SIDECAR_INSTANCE_ID, "-1");
+ assertThatThrownBy(() -> bulkSparkConf.getSidecarInstanceId())
+ .isExactlyInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Spark conf " + BulkSparkConf.SIDECAR_INSTANCE_ID
+ + " must be a non-negative integer; got -1");
+ }
+
+ @Test
+ void testSidecarInstanceIdNonIntegerThrows()
+ {
+ sparkConf.set(BulkSparkConf.SIDECAR_INSTANCE_ID, "notanint");
+ assertThatThrownBy(() -> bulkSparkConf.getSidecarInstanceId())
+ .isExactlyInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Spark conf " + BulkSparkConf.SIDECAR_INSTANCE_ID
+ + " is not set to a valid integer string");
+ }
+
@Test
void testDefaultSidecarPort()
{
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfoTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfoTest.java
index 99b06712f..6f101e319 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfoTest.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/CassandraClusterInfoTest.java
@@ -21,14 +21,19 @@
import java.time.Duration;
import java.time.Instant;
+import java.util.Map;
+import java.util.Set;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Maps;
import com.google.common.util.concurrent.Uninterruptibles;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
@@ -36,12 +41,18 @@
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
+import o.a.c.sidecar.client.shaded.client.SidecarInstance;
import o.a.c.sidecar.client.shaded.common.response.NodeSettings;
import o.a.c.sidecar.client.shaded.common.response.TimeSkewResponse;
+import o.a.c.sidecar.client.shaded.common.response.data.RingEntry;
import org.apache.cassandra.spark.bulkwriter.token.TokenRangeMapping;
+import org.apache.cassandra.spark.common.SidecarInstanceFactory;
import org.apache.cassandra.spark.exception.TimeSkewTooLargeException;
+import org.apache.spark.SparkConf;
+import org.jetbrains.annotations.Nullable;
import static org.apache.cassandra.spark.TestUtils.range;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatNoException;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
@@ -120,6 +131,153 @@ public static CassandraClusterInfo mockClusterInfoForTimeSkewTest(int allowanceM
return new MockClusterInfoForTimeSkew(allowanceMinutes, remoteNow);
}
+ @Test
+ void testValidateSidecarInstanceIdCoverageThrowsWhenPartiallyConfigured()
+ {
+ CassandraClusterInfo ci = noOpClusterInfoWithGlobalInstanceId(1);
+ Set instances = new HashSet<>();
+ instances.add(ringInstance("dc1-i0", 1));
+ instances.add(ringInstance("dc1-i1", null));
+ instances.add(ringInstance("dc1-i2", null));
+
+ assertThatThrownBy(() -> ci.validateSidecarInstanceIdCoverage(instances))
+ .describedAs("2/3 instances would fall back to the global id and collide on whichever instance it identifies")
+ .isExactlyInstanceOf(IllegalStateException.class)
+ .hasMessageContaining(BulkSparkConf.SIDECAR_INSTANCE_ID + "=1")
+ .hasMessageContaining("dc1-i1")
+ .hasMessageContaining("dc1-i2");
+ }
+
+ @Test
+ void testValidateSidecarInstanceIdCoverageAllowsFullCoverage()
+ {
+ CassandraClusterInfo ci = noOpClusterInfoWithGlobalInstanceId(1);
+ Set instances = new HashSet<>();
+ instances.add(ringInstance("dc1-i0", 1));
+ instances.add(ringInstance("dc1-i1", 2));
+ instances.add(ringInstance("dc1-i2", 3));
+
+ assertThatNoException()
+ .describedAs("every instance resolves its own id, so the global id is never actually used")
+ .isThrownBy(() -> ci.validateSidecarInstanceIdCoverage(instances));
+ }
+
+ @Test
+ void testValidateSidecarInstanceIdCoverageNoopForSingleInstance()
+ {
+ CassandraClusterInfo ci = noOpClusterInfoWithGlobalInstanceId(1);
+ Set instances = Collections.singleton(ringInstance("dc1-i0", null));
+
+ assertThatNoException()
+ .describedAs("a single instance is unambiguous: the global id can only apply to it")
+ .isThrownBy(() -> ci.validateSidecarInstanceIdCoverage(instances));
+ }
+
+ @Test
+ void testValidateSidecarInstanceIdCoverageNoopWhenGlobalIdUnset()
+ {
+ CassandraClusterInfo ci = noOpClusterInfoWithGlobalInstanceId(null);
+ Set instances = new HashSet<>();
+ instances.add(ringInstance("dc1-i0", null));
+ instances.add(ringInstance("dc1-i1", null));
+
+ assertThatNoException().isThrownBy(() -> ci.validateSidecarInstanceIdCoverage(instances));
+ }
+
+ @Test
+ void testSidecarInstanceIdsByHostnameFromPlainContactPoints()
+ {
+ Set contactPoints = new HashSet<>(Arrays.asList(
+ SidecarInstanceFactory.createFromString("cassandra1:9043=1", 9043),
+ SidecarInstanceFactory.createFromString("cassandra2:9043=2", 9043),
+ SidecarInstanceFactory.createFromString("cassandra3:9043", 9043)));
+
+ Map byHostname = CassandraClusterInfo.sidecarInstanceIdsByHostname(contactPoints);
+
+ assertThat(byHostname).containsEntry("cassandra1", 1).containsEntry("cassandra2", 2);
+ assertThat(byHostname)
+ .describedAs("contact point with no '=' suffix has no entry")
+ .doesNotContainKey("cassandra3");
+ }
+
+ @Test
+ void testSidecarInstanceIdsByHostnameFromCoordinatedWriteContactPoints()
+ {
+ // Coordinated-write contact points are parsed the same way (SimpleClusterConf.buildSidecarContactPoints
+ // also delegates to SidecarInstanceFactory.createFromString), so per-instance ids must resolve here too -
+ // this is the exact path that regressed when id lookup was previously sourced from
+ // conf.sidecarContactPoints() instead of the cluster's actually-resolved contact points.
+ Set coordinatedContactPoints = new HashSet<>(Arrays.asList(
+ SidecarInstanceFactory.createFromString("172.20.39.166:9043=1"),
+ SidecarInstanceFactory.createFromString("172.20.39.97:9043=2"),
+ SidecarInstanceFactory.createFromString("172.20.39.216:9043=3")));
+
+ Map byHostname = CassandraClusterInfo.sidecarInstanceIdsByHostname(coordinatedContactPoints);
+
+ assertThat(byHostname).containsEntry("172.20.39.166", 1)
+ .containsEntry("172.20.39.97", 2)
+ .containsEntry("172.20.39.216", 3);
+ }
+
+ @Test
+ void testSidecarInstanceIdsByHostnameEmptyWhenNoneConfigured()
+ {
+ Set contactPoints = new HashSet<>(Arrays.asList(
+ SidecarInstanceFactory.createFromString("cassandra1:9043", 9043),
+ SidecarInstanceFactory.createFromString("cassandra2:9043", 9043)));
+
+ assertThat(CassandraClusterInfo.sidecarInstanceIdsByHostname(contactPoints)).isEmpty();
+ }
+
+ private static RingInstance ringInstance(String fqdn, @Nullable Integer sidecarInstanceId)
+ {
+ return new RingInstance(new RingEntry.Builder()
+ .datacenter("dc1")
+ .address(fqdn)
+ .port(7000)
+ .status("UP")
+ .state("NORMAL")
+ .token("0")
+ .fqdn(fqdn)
+ .rack("rack")
+ .owns("")
+ .load("")
+ .hostId("")
+ .build(), null, sidecarInstanceId);
+ }
+
+ private static CassandraClusterInfo noOpClusterInfoWithGlobalInstanceId(@Nullable Integer globalInstanceId)
+ {
+ Map options = Maps.newTreeMap(String.CASE_INSENSITIVE_ORDER);
+ options.put(WriterOptions.SIDECAR_CONTACT_POINTS.name(), "127.0.0.1");
+ options.put(WriterOptions.KEYSPACE.name(), "ks");
+ options.put(WriterOptions.TABLE.name(), "table");
+ options.put(WriterOptions.KEYSTORE_PASSWORD.name(), "dummy_password");
+ options.put(WriterOptions.KEYSTORE_BASE64_ENCODED.name(), "ZHVtbXk=");
+
+ SparkConf sparkConf = new SparkConf();
+ if (globalInstanceId != null)
+ {
+ sparkConf.set(BulkSparkConf.SIDECAR_INSTANCE_ID, globalInstanceId.toString());
+ }
+ return new NoOpClusterInfo(new BulkSparkConf(sparkConf, options));
+ }
+
+ private static class NoOpClusterInfo extends CassandraClusterInfo
+ {
+ NoOpClusterInfo(BulkSparkConf conf)
+ {
+ super(conf);
+ }
+
+ @Override
+ protected CassandraContext buildCassandraContext()
+ {
+ CassandraContext context = mock(CassandraContext.class, RETURNS_DEEP_STUBS);
+ when(context.getCluster()).thenReturn(Collections.emptySet());
+ return context;
+ }
+
private BulkSparkConf mockBulkSparkWithSidecarConf(int requestTimeoutSeconds, long maxRetryDelayMillis, int retryCount)
{
BulkSparkConf conf = mock(BulkSparkConf.class);
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RingInstanceSerializationTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RingInstanceSerializationTest.java
index c2ff45901..85ad5fcbc 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RingInstanceSerializationTest.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RingInstanceSerializationTest.java
@@ -75,4 +75,28 @@ public void testRingSerializesFromReplicaMetadata()
RingInstance deserialized = deserialize(bytes, RingInstance.class);
assertThat(deserialized).isEqualTo(ring);
}
+
+ @Test
+ public void testSidecarInstanceIdSurvivesSerialization()
+ {
+ int dcOffset = 0;
+ String dataCenter = "DC1";
+ int index = 0;
+ ReplicaMetadata metadata = new ReplicaMetadata("NORMAL",
+ "UP",
+ dataCenter + "-i" + index,
+ "127.0." + dcOffset + "." + index,
+ 7000,
+ dataCenter);
+
+ RingInstance ring = new RingInstance(metadata, "test-cluster", 2);
+
+ byte[] bytes = serialize(ring);
+ RingInstance deserialized = deserialize(bytes, RingInstance.class);
+ // sidecarInstanceId is excluded from equals/hashCode, so it must be checked explicitly:
+ // this is exactly the field that would silently revert to null if serialization dropped it,
+ // which would reintroduce the misrouting bug once the instance travels to an executor.
+ assertThat(deserialized).isEqualTo(ring);
+ assertThat(deserialized.sidecarInstanceId()).isEqualTo(2);
+ }
}
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RingInstanceTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RingInstanceTest.java
index b73cc272a..0aecd7157 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RingInstanceTest.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RingInstanceTest.java
@@ -178,16 +178,42 @@ public void testToString()
{
RingEntry ringEntry = mockRingEntry();
RingInstance instanceWithoutClusterId = new RingInstance(ringEntry);
- assertThat(instanceWithoutClusterId.toString()).isEqualTo("RingInstance{cluster='null', " +
+ assertThat(instanceWithoutClusterId.toString()).isEqualTo("RingInstance{cluster='null', sidecarInstanceId=null, " +
"RingEntry{datacenter='DATACENTER1', address='127.0.0.1', port=0, rack='Rack', " +
"status='UP', state='NORMAL', load='0', owns='', token='0', fqdn='DATACENTER1-i1', hostId=''}}");
RingInstance instanceWithClusterId = new RingInstance(ringEntry, "clusterId");
- assertThat(instanceWithClusterId.toString()).isEqualTo("RingInstance{cluster='clusterId', " +
+ assertThat(instanceWithClusterId.toString()).isEqualTo("RingInstance{cluster='clusterId', sidecarInstanceId=null, " +
"RingEntry{datacenter='DATACENTER1', address='127.0.0.1', port=0, rack='Rack', " +
"status='UP', state='NORMAL', load='0', owns='', token='0', fqdn='DATACENTER1-i1', hostId=''}}");
}
+ @Test
+ public void testSidecarInstanceIdDefaultsToNull()
+ {
+ RingInstance instance = new RingInstance(mockRingEntry());
+ assertThat(instance.sidecarInstanceId()).isNull();
+ }
+
+ @Test
+ public void testSidecarInstanceIdIsRetained()
+ {
+ RingInstance instance = new RingInstance(mockRingEntry(), null, 2);
+ assertThat(instance.sidecarInstanceId()).isEqualTo(2);
+ }
+
+ @Test
+ public void testEqualsAndHashcodeIgnoreSidecarInstanceId()
+ {
+ RingEntry ringEntry = mockRingEntry();
+ RingInstance instanceWithId = new RingInstance(ringEntry, null, 1);
+ RingInstance instanceWithDifferentId = new RingInstance(ringEntry, null, 2);
+ RingInstance instanceWithoutId = new RingInstance(ringEntry);
+
+ assertThat(instanceWithId).isEqualTo(instanceWithDifferentId).isEqualTo(instanceWithoutId);
+ assertThat(instanceWithId.hashCode()).isEqualTo(instanceWithDifferentId.hashCode()).isEqualTo(instanceWithoutId.hashCode());
+ }
+
@NotNull
private static RingEntry mockRingEntry()
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/SidecarDataTransferApiTest.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/SidecarDataTransferApiTest.java
new file mode 100644
index 000000000..f2d4baeea
--- /dev/null
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/SidecarDataTransferApiTest.java
@@ -0,0 +1,88 @@
+/*
+ * 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.spark.bulkwriter;
+
+import org.junit.jupiter.api.Test;
+
+import o.a.c.sidecar.client.shaded.client.SidecarInstanceImpl;
+import o.a.c.sidecar.client.shaded.common.response.data.RingEntry;
+import org.apache.cassandra.bridge.CassandraBridge;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link SidecarDataTransferApi}
+ */
+class SidecarDataTransferApiTest
+{
+ @Test
+ void testToSidecarInstanceCarriesPerInstanceId()
+ {
+ SidecarDataTransferApi api = api();
+ RingInstance instance = ringInstance("dc1-i0", 2);
+
+ SidecarInstanceImpl sidecarInstance = api.toSidecarInstance(instance);
+
+ assertThat(sidecarInstance.hostname()).isEqualTo("dc1-i0");
+ assertThat(sidecarInstance.port()).isEqualTo(9043);
+ assertThat(sidecarInstance.instanceId())
+ .describedAs("upload/commit/cleanup requests must carry the target instance's own id, "
+ + "not a single job-wide value, or a multi-node write silently misroutes")
+ .isEqualTo(2);
+ }
+
+ @Test
+ void testToSidecarInstanceFallsBackToNullWhenNoPerInstanceIdConfigured()
+ {
+ SidecarDataTransferApi api = api();
+ RingInstance instance = ringInstance("dc1-i1", null);
+
+ SidecarInstanceImpl sidecarInstance = api.toSidecarInstance(instance);
+
+ assertThat(sidecarInstance.instanceId()).isNull();
+ }
+
+ private static SidecarDataTransferApi api()
+ {
+ CassandraContext context = mock(CassandraContext.class, RETURNS_DEEP_STUBS);
+ when(context.sidecarPort()).thenReturn(9043);
+ return new SidecarDataTransferApi(context, mock(CassandraBridge.class), mock(JobInfo.class));
+ }
+
+ private static RingInstance ringInstance(String fqdn, Integer sidecarInstanceId)
+ {
+ return new RingInstance(new RingEntry.Builder()
+ .datacenter("dc1")
+ .address(fqdn)
+ .port(7000)
+ .status("UP")
+ .state("NORMAL")
+ .token("0")
+ .fqdn(fqdn)
+ .rack("rack")
+ .owns("")
+ .load("")
+ .hostId("")
+ .build(), null, sidecarInstanceId);
+ }
+}
diff --git a/cassandra-analytics-core/src/test/spark3/org/apache/cassandra/spark/common/SidecarInstanceFactoryTest.java b/cassandra-analytics-core/src/test/spark3/org/apache/cassandra/spark/common/SidecarInstanceFactoryTest.java
index 277175a7f..6c8a6663c 100644
--- a/cassandra-analytics-core/src/test/spark3/org/apache/cassandra/spark/common/SidecarInstanceFactoryTest.java
+++ b/cassandra-analytics-core/src/test/spark3/org/apache/cassandra/spark/common/SidecarInstanceFactoryTest.java
@@ -47,9 +47,46 @@ void testCreateSidecarInstance()
"[2024:a::1]", 8888);
}
+ @Test
+ void testCreateSidecarInstanceWithInstanceId()
+ {
+ assertSidecarInstance(SidecarInstanceFactory.createFromString("localhost:8888=2", 9999),
+ "localhost", 8888, 2);
+ assertSidecarInstance(SidecarInstanceFactory.createFromString("127.0.0.1:8888=0", 9999),
+ "127.0.0.1", 8888, 0);
+ // no explicit port: default port applies, id still parsed
+ assertSidecarInstance(SidecarInstanceFactory.createFromString("localhost=3", 9999),
+ "localhost", 9999, 3);
+ // ipv6 with port and id
+ assertSidecarInstance(SidecarInstanceFactory.createFromString("[2024:a::1]:8888=7", 9999),
+ "[2024:a::1]", 8888, 7);
+ // no id: instanceId is null (falls back to the job-level value)
+ assertSidecarInstance(SidecarInstanceFactory.createFromString("localhost:8888", 9999),
+ "localhost", 8888, null);
+ }
+
+ @Test
+ void testCreateSidecarInstanceWithInvalidInstanceId()
+ {
+ assertThatThrownBy(() -> SidecarInstanceFactory.createFromString("localhost:8888=abc", 9999))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Invalid sidecar instanceId");
+
+ assertThatThrownBy(() -> SidecarInstanceFactory.createFromString("localhost:8888=-1", 9999))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("non-negative");
+ }
+
private void assertSidecarInstance(SidecarInstance sidecarInstance, String expectedHostname, int expectedPort)
{
assertThat(sidecarInstance.hostname()).isEqualTo(expectedHostname);
assertThat(sidecarInstance.port()).isEqualTo(expectedPort);
}
+
+ private void assertSidecarInstance(SidecarInstance sidecarInstance, String expectedHostname, int expectedPort,
+ Integer expectedInstanceId)
+ {
+ assertSidecarInstance(sidecarInstance, expectedHostname, expectedPort);
+ assertThat(sidecarInstance.instanceId()).isEqualTo(expectedInstanceId);
+ }
}
diff --git a/cassandra-analytics-core/src/test/spark4/org/apache/cassandra/spark/common/SidecarInstanceFactoryTest.java b/cassandra-analytics-core/src/test/spark4/org/apache/cassandra/spark/common/SidecarInstanceFactoryTest.java
index 277175a7f..6c8a6663c 100644
--- a/cassandra-analytics-core/src/test/spark4/org/apache/cassandra/spark/common/SidecarInstanceFactoryTest.java
+++ b/cassandra-analytics-core/src/test/spark4/org/apache/cassandra/spark/common/SidecarInstanceFactoryTest.java
@@ -47,9 +47,46 @@ void testCreateSidecarInstance()
"[2024:a::1]", 8888);
}
+ @Test
+ void testCreateSidecarInstanceWithInstanceId()
+ {
+ assertSidecarInstance(SidecarInstanceFactory.createFromString("localhost:8888=2", 9999),
+ "localhost", 8888, 2);
+ assertSidecarInstance(SidecarInstanceFactory.createFromString("127.0.0.1:8888=0", 9999),
+ "127.0.0.1", 8888, 0);
+ // no explicit port: default port applies, id still parsed
+ assertSidecarInstance(SidecarInstanceFactory.createFromString("localhost=3", 9999),
+ "localhost", 9999, 3);
+ // ipv6 with port and id
+ assertSidecarInstance(SidecarInstanceFactory.createFromString("[2024:a::1]:8888=7", 9999),
+ "[2024:a::1]", 8888, 7);
+ // no id: instanceId is null (falls back to the job-level value)
+ assertSidecarInstance(SidecarInstanceFactory.createFromString("localhost:8888", 9999),
+ "localhost", 8888, null);
+ }
+
+ @Test
+ void testCreateSidecarInstanceWithInvalidInstanceId()
+ {
+ assertThatThrownBy(() -> SidecarInstanceFactory.createFromString("localhost:8888=abc", 9999))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Invalid sidecar instanceId");
+
+ assertThatThrownBy(() -> SidecarInstanceFactory.createFromString("localhost:8888=-1", 9999))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("non-negative");
+ }
+
private void assertSidecarInstance(SidecarInstance sidecarInstance, String expectedHostname, int expectedPort)
{
assertThat(sidecarInstance.hostname()).isEqualTo(expectedHostname);
assertThat(sidecarInstance.port()).isEqualTo(expectedPort);
}
+
+ private void assertSidecarInstance(SidecarInstance sidecarInstance, String expectedHostname, int expectedPort,
+ Integer expectedInstanceId)
+ {
+ assertSidecarInstance(sidecarInstance, expectedHostname, expectedPort);
+ assertThat(sidecarInstance.instanceId()).isEqualTo(expectedInstanceId);
+ }
}
diff --git a/cassandra-analytics-sidecar-client/src/main/java/org/apache/cassandra/clients/Sidecar.java b/cassandra-analytics-sidecar-client/src/main/java/org/apache/cassandra/clients/Sidecar.java
index 78798416a..1a9269c27 100644
--- a/cassandra-analytics-sidecar-client/src/main/java/org/apache/cassandra/clients/Sidecar.java
+++ b/cassandra-analytics-sidecar-client/src/main/java/org/apache/cassandra/clients/Sidecar.java
@@ -235,7 +235,7 @@ else if (gossipInfoResponses.size() < gossipInfoFutures.size())
public static SidecarInstance toSidecarInstance(CassandraInstance instance, int sidecarPort)
{
- return new SidecarInstanceImpl(instance.nodeName(), sidecarPort);
+ return new SidecarInstanceImpl(instance.nodeName(), sidecarPort, instance.sidecarInstanceId());
}
public static final class ClientConfig