From 9bf1a4fd3966cad593952fe1244c36db610cb497 Mon Sep 17 00:00:00 2001 From: Bianca Stanciu Date: Thu, 9 Jul 2026 12:47:54 +0300 Subject: [PATCH 1/2] CASSANALYTICS-177: Add instanceId query parameter to fix HTTP 421 errors when Sidecar is behind a load balancer --- CHANGES.txt | 1 + .../common/http/SidecarQueryParamNames.java | 40 +++++++++++ .../sidecar/client/HttpClientConfig.java | 33 +++++++++ .../sidecar/client/HttpClientConfigTest.java | 37 ++++++++++ .../sidecar/client/VertxHttpClient.java | 8 +++ .../sidecar/client/VertxHttpClientTest.java | 57 ++++++++++++++-- .../clients/AnalyticsSidecarClient.java | 44 ++++++++---- .../spark/bulkwriter/BulkSparkConf.java | 16 +++++ .../clients/AnalyticsSidecarClientTest.java | 67 +++++++++++++++++++ .../spark/bulkwriter/BulkSparkConfTest.java | 40 +++++++++++ 10 files changed, 325 insertions(+), 18 deletions(-) create mode 100644 analytics-sidecar-client-common/src/main/java/org/apache/cassandra/sidecar/common/http/SidecarQueryParamNames.java create mode 100644 cassandra-analytics-core/src/test/java/org/apache/cassandra/clients/AnalyticsSidecarClientTest.java diff --git a/CHANGES.txt b/CHANGES.txt index 974b9ceaa..297fcbf41 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) * 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) 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-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..f4e455dec 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,13 @@ protected HttpRequest 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 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..8d1a4a125 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,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 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"); + } + } + + 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<>() 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..d198a5abd 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,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) 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 a01c2d1e3..02b0e92ad 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 @@ -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); 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..641df2432 --- /dev/null +++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/clients/AnalyticsSidecarClientTest.java @@ -0,0 +1,67 @@ +/* + * 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.Map; + +import com.google.common.collect.Maps; +import org.junit.jupiter.api.Test; + +import o.a.c.sidecar.client.shaded.client.HttpClientConfig; +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; + +/** + * 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); + } + + 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 944f0a186..24a7ea5d1 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() { From 0493eb3fd66560d6184286c481ebc36d99196d93 Mon Sep 17 00:00:00 2001 From: Bianca Stanciu Date: Wed, 12 Aug 2026 18:28:59 +0300 Subject: [PATCH 2/2] CASSANALYTICS-177: Resolve instanceId per sidecar instance to prevent misrouting on multi-node bulk write jobs --- .../sidecar/client/SidecarInstance.java | 16 ++ .../sidecar/client/SidecarInstanceImpl.java | 39 ++++- .../client/SidecarInstanceImplTest.java | 40 +++++ .../sidecar/client/VertxHttpClient.java | 13 +- .../sidecar/client/VertxHttpClientTest.java | 31 ++++ .../spark/common/model/CassandraInstance.java | 16 ++ .../clients/AnalyticsSidecarClient.java | 44 ++++- .../bulkwriter/CassandraClusterInfo.java | 70 +++++++- .../spark/bulkwriter/RingInstance.java | 38 ++++- .../bulkwriter/SidecarDataTransferApi.java | 2 +- .../spark/common/SidecarInstanceFactory.java | 39 ++++- .../clients/AnalyticsSidecarClientTest.java | 65 +++++++ .../bulkwriter/CassandraClusterInfoTest.java | 159 ++++++++++++++++++ .../RingInstanceSerializationTest.java | 24 +++ .../spark/bulkwriter/RingInstanceTest.java | 30 +++- .../SidecarDataTransferApiTest.java | 88 ++++++++++ .../common/SidecarInstanceFactoryTest.java | 37 ++++ .../common/SidecarInstanceFactoryTest.java | 37 ++++ .../org/apache/cassandra/clients/Sidecar.java | 2 +- 19 files changed, 759 insertions(+), 31 deletions(-) create mode 100644 cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/SidecarDataTransferApiTest.java 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/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 f4e455dec..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 @@ -253,11 +253,16 @@ protected HttpRequest vertxRequest(SidecarInstance sidecarInstance, Requ sidecarInstance.hostname(), request.requestURI()); - if (config.instanceId() != null) + // 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(config.instanceId())); - LOGGER.debug("Appended {}={} to request uri. originalUri={}, finalUri={}", - INSTANCE_ID, config.instanceId(), request.requestURI(), vertxRequest.uri()); + 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()); 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 8d1a4a125..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 @@ -107,11 +107,42 @@ public void testInstanceIdQueryParamAppendedWithExistingQueryParams() } } + @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; } 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 d198a5abd..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,6 +19,8 @@ package org.apache.cassandra.clients; +import java.util.List; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -28,6 +30,7 @@ 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; @@ -51,11 +54,7 @@ public static SidecarClient from(SidecarInstancesProvider sidecarInstancesProvid .setWorkerPoolSize(conf.getMaxHttpConnections())); HttpClientConfig httpClientConfig = buildHttpClientConfig(conf); - if (httpClientConfig.instanceId() != null) - { - LOGGER.info("Sidecar HTTP client configured with instanceId={} (applied to every outbound sidecar request)", - httpClientConfig.instanceId()); - } + warnIfGlobalInstanceIdIsAmbiguous(httpClientConfig, sidecarInstancesProvider); StartupValidator.instance().register(new SslValidation(conf)); StartupValidator.instance().register(new BulkWriterKeyStoreValidation(conf)); @@ -71,6 +70,41 @@ 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()); 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 1b1a40179..f78abf5ff 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 index 641df2432..e70bc0d08 100644 --- 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 @@ -19,17 +19,22 @@ 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} @@ -53,6 +58,66 @@ void testBuildHttpClientConfigWiresInstanceId() 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); 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 5ed4c5221..c3473eddf 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,17 +21,28 @@ import java.time.Duration; import java.time.Instant; +import java.util.Arrays; import java.util.Collections; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; import java.util.concurrent.CompletableFuture; import com.google.common.collect.ImmutableMap; +import com.google.common.collect.Maps; import org.junit.jupiter.api.Test; +import o.a.c.sidecar.client.shaded.client.SidecarInstance; 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; @@ -75,6 +86,154 @@ 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 static class MockClusterInfoForTimeSkew extends CassandraClusterInfo { private CassandraContext cassandraContext; 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