content: /*
   * 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.client;
  
  import java.io.ByteArrayOutputStream;
  import java.io.IOException;
  import java.io.InputStream;
  import java.nio.charset.Charset;
  import java.nio.charset.StandardCharsets;
  import java.nio.file.Files;
  import java.nio.file.Path;
  import java.nio.file.StandardCopyOption;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.UUID;
  import java.util.concurrent.CompletableFuture;
  import java.util.concurrent.CountDownLatch;
  import java.util.concurrent.ExecutionException;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.TimeoutException;
  import java.util.function.Consumer;
  import java.util.stream.Collectors;
  
  import org.junit.jupiter.api.AfterEach;
  import org.junit.jupiter.api.BeforeEach;
  import org.junit.jupiter.api.Test;
  import org.junit.jupiter.api.io.TempDir;
  import org.junit.jupiter.params.ParameterizedTest;
  import org.junit.jupiter.params.provider.ValueSource;
  
  import com.fasterxml.jackson.core.JsonProcessingException;
  import com.fasterxml.jackson.databind.ObjectMapper;
  import io.netty.handler.codec.http.HttpHeaderNames;
  import io.netty.handler.codec.http.HttpHeaderValues;
  import okhttp3.mockwebserver.MockResponse;
  import okhttp3.mockwebserver.MockWebServer;
  import okhttp3.mockwebserver.RecordedRequest;
  import okhttp3.mockwebserver.SocketPolicy;
  import okio.Buffer;
  import okio.Okio;
  import org.apache.cassandra.sidecar.client.exception.RetriesExhaustedException;
  import org.apache.cassandra.sidecar.client.request.RequestExecutorTest;
  import org.apache.cassandra.sidecar.client.retry.RetryAction;
  import org.apache.cassandra.sidecar.client.retry.RetryPolicy;
  import org.apache.cassandra.sidecar.common.ApiEndpointsV1;
  import org.apache.cassandra.sidecar.common.data.Lifecycle.CassandraState;
  import org.apache.cassandra.sidecar.common.data.Lifecycle.OperationStatus;
  import org.apache.cassandra.sidecar.common.data.OperationalJobStatus;
  import org.apache.cassandra.sidecar.common.data.RestoreJobSecrets;
  import org.apache.cassandra.sidecar.common.request.ImportSSTableRequest;
  import org.apache.cassandra.sidecar.common.request.NodeSettingsRequest;
  import org.apache.cassandra.sidecar.common.request.Request;
  import org.apache.cassandra.sidecar.common.request.Service;
  import org.apache.cassandra.sidecar.common.request.data.AllServicesConfigPayload;
  import org.apache.cassandra.sidecar.common.request.data.CreateRestoreJobRequestPayload;
  import org.apache.cassandra.sidecar.common.request.data.MD5Digest;
  import org.apache.cassandra.sidecar.common.request.data.NodeCommandRequestPayload;
  import org.apache.cassandra.sidecar.common.request.data.UpdateCdcServiceConfigPayload;
  import org.apache.cassandra.sidecar.common.request.data.XXHash32Digest;
  import org.apache.cassandra.sidecar.common.response.CompactionStatsResponse;
  import org.apache.cassandra.sidecar.common.response.ConnectedClientStatsResponse;
  import org.apache.cassandra.sidecar.common.response.GossipInfoResponse;
  import org.apache.cassandra.sidecar.common.response.HealthResponse;
  import org.apache.cassandra.sidecar.common.response.InstanceFileInfo;
  import org.apache.cassandra.sidecar.common.response.InstanceFilesListResponse;
  import org.apache.cassandra.sidecar.common.response.LifecycleInfoResponse;
  import org.apache.cassandra.sidecar.common.response.ListCdcSegmentsResponse;
  import org.apache.cassandra.sidecar.common.response.ListOperationalJobsResponse;
  import org.apache.cassandra.sidecar.common.response.ListSnapshotFilesResponse;
  import org.apache.cassandra.sidecar.common.response.NodeSettings;
  import org.apache.cassandra.sidecar.common.response.OperationalJobResponse;
  import org.apache.cassandra.sidecar.common.response.RingResponse;
  import org.apache.cassandra.sidecar.common.response.SSTableImportResponse;
  import org.apache.cassandra.sidecar.common.response.SchemaResponse;
  import org.apache.cassandra.sidecar.common.response.StreamStatsResponse;
  import org.apache.cassandra.sidecar.common.response.TableStatsResponse;
  import org.apache.cassandra.sidecar.common.response.TimeSkewResponse;
  import org.apache.cassandra.sidecar.common.response.TokenRangeReplicasResponse;
  import org.apache.cassandra.sidecar.common.response.data.CdcSegmentInfo;
  import org.apache.cassandra.sidecar.common.response.data.ClientConnectionEntry;
  import org.apache.cassandra.sidecar.common.response.data.CreateRestoreJobResponsePayload;
  import org.apache.cassandra.sidecar.common.response.data.RingEntry;
  import org.apache.cassandra.sidecar.common.response.data.StreamsProgressStats;
  import org.apache.cassandra.sidecar.common.utils.HttpRange;
  import org.apache.cassandra.sidecar.foundation.RestoreJobSecretsGen;
  
  import static io.netty.handler.codec.http.HttpResponseStatus.ACCEPTED;
  import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;
  import static io.netty.handler.codec.http.HttpResponseStatus.INTERNAL_SERVER_ERROR;
  import static io.netty.handler.codec.http.HttpResponseStatus.OK;
  import static io.netty.handler.codec.http.HttpResponseStatus.PARTIAL_CONTENT;
  import static org.apache.cassandra.sidecar.common.ApiEndpointsV1.JOB_ID_PATH_PARAM;
  import static org.apache.cassandra.sidecar.common.ApiEndpointsV1.KEYSPACE_PATH_PARAM;
  import static org.apache.cassandra.sidecar.common.ApiEndpointsV1.LIVE_MIGRATION_FILES_ROUTE;
  import static org.apache.cassandra.sidecar.common.ApiEndpointsV1.OPERATIONAL_JOB_ID_PATH_PARAM;
  import static org.apache.cassandra.sidecar.common.ApiEndpointsV1.TABLE_PATH_PARAM;
  import static org.apache.cassandra.sidecar.common.http.SidecarHttpHeaderNames.CONTENT_XXHASH32;
  import static org.apache.cassandra.sidecar.common.http.SidecarHttpHeaderNames.CONTENT_XXHASH32_SEED;
  import static org.apache.cassandra.sidecar.common.response.InstanceFileInfo.FileType.DIRECTORY;
  import static org.apache.cassandra.sidecar.common.response.InstanceFileInfo.FileType.FILE;
  import static org.assertj.core.api.Assertions.assertThat;
  import static org.assertj.core.api.Assertions.assertThatException;
  import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
  import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
  import static org.assertj.core.api.Assertions.assertThatNoException;
  import static org.assertj.core.api.Assertions.assertThatThrownBy;
  import static org.assertj.core.api.Assertions.fail;
  
  abstract class SidecarClientTest
  {
      SidecarClient client;
      List<MockWebServer> servers;
      List<SidecarInstanceImpl> instances;
  
      @BeforeEach
      void setup()
      {
          servers = new ArrayList<>();
          for (int i = 0; i < 4; i++)
          {
              servers.add(new MockWebServer());
          }
  
          instances = servers.stream()
                             .map(RequestExecutorTest::newSidecarInstance)
                             .collect(Collectors.toList());
          client = initialize(instances);
      }
  
      protected abstract SidecarClient initialize(List<SidecarInstanceImpl> instances);
  
      @AfterEach
      void tearDown() throws Exception
      {
          for (MockWebServer server : servers)
          {
              server.shutdown();
          }
          client.close();
      }
  
      @Test
      void testSidecarHealthOk() throws Exception
      {
          MockResponse response = new MockResponse()
                                  .setResponseCode(200)
                                  .setHeader("content-type", "application/json")
                                  .setBody("{\"status\":\"OK\"}");
          enqueue(response);
  
          HealthResponse result = client.sidecarHealth().get(30, TimeUnit.SECONDS);
          assertThat(result).isNotNull();
          assertThat(result.status()).isEqualToIgnoringCase("OK");
          assertThat(result.isOk()).isTrue();
  
          validateResponseServed(ApiEndpointsV1.HEALTH_ROUTE);
      }
  
      @Test
      void testSidecarHealthNotOk() throws Exception
      {
          MockResponse response = new MockResponse()
                                  .setResponseCode(503)
                                  .setHeader("content-type", "application/json")
                                  .setBody("{\"status\":\"NOT_OK\"}");
          enqueue(response);
  
          assertThatThrownBy(() -> client.sidecarHealth().get(30, TimeUnit.SECONDS))
          .isInstanceOf(ExecutionException.class)
          .hasCauseInstanceOf(RetriesExhaustedException.class);
  
          validateResponseServed(ApiEndpointsV1.HEALTH_ROUTE);
      }
  
      @SuppressWarnings("deprecation")
      @Test
      void testCassandraDeprecatedHealthOk() throws Exception
      {
          MockResponse response = new MockResponse()
                                  .setResponseCode(200)
                                  .setHeader("content-type", "application/json")
                                  .setBody("{\"status\":\"OK\"}");
          enqueue(response);
  
          HealthResponse result = client.cassandraHealth().get(30, TimeUnit.SECONDS);
          assertThat(result).isNotNull();
          assertThat(result.status()).isEqualToIgnoringCase("OK");
          assertThat(result.isOk()).isTrue();
  
          validateResponseServed(ApiEndpointsV1.CASSANDRA_HEALTH_ROUTE);
      }
  
      @SuppressWarnings("deprecation")
      @Test
      void testCassandraDeprecatedHealthNotOk() throws Exception
      {
          MockResponse response = new MockResponse()
                                  .setResponseCode(503)
                                  .setHeader("content-type", "application/json")
                                  .setBody("{\"status\":\"NOT_OK\"}");
          enqueue(response);
  
          assertThatThrownBy(() -> client.cassandraHealth().get(30, TimeUnit.SECONDS))
          .isInstanceOf(ExecutionException.class)
          .hasCauseInstanceOf(RetriesExhaustedException.class);
  
          validateResponseServed(ApiEndpointsV1.CASSANDRA_HEALTH_ROUTE);
      }
  
      @Test
      void testCassandraNativeHealthOk() throws Exception
      {
          MockResponse response = new MockResponse()
                                  .setResponseCode(200)
                                  .setHeader("content-type", "application/json")
                                  .setBody("{\"status\":\"OK\"}");
          enqueue(response);
  
          HealthResponse result = client.cassandraNativeHealth().get(30, TimeUnit.SECONDS);
          assertThat(result).isNotNull();
          assertThat(result.status()).isEqualToIgnoringCase("OK");
          assertThat(result.isOk()).isTrue();
  
          validateResponseServed(ApiEndpointsV1.CASSANDRA_NATIVE_HEALTH_ROUTE);
      }
  
      @Test
      void testCassandraNativeHealthNotOk() throws Exception
      {
          MockResponse response = new MockResponse()
                                  .setResponseCode(503)
                                  .setHeader("content-type", "application/json")
                                  .setBody("{\"status\":\"NOT_OK\"}");
          enqueue(response);
  
          assertThatThrownBy(() -> client.cassandraNativeHealth().get(30, TimeUnit.SECONDS))
          .isInstanceOf(ExecutionException.class)
          .hasCauseInstanceOf(RetriesExhaustedException.class);
  
          validateResponseServed(ApiEndpointsV1.CASSANDRA_NATIVE_HEALTH_ROUTE);
      }
  
      @Test
      void testCassandraJmxHealthOk() throws Exception
      {
          MockResponse response = new MockResponse()
                                  .setResponseCode(200)
                                  .setHeader("content-type", "application/json")
                                  .setBody("{\"status\":\"OK\"}");
          enqueue(response);
  
          HealthResponse result = client.cassandraJmxHealth().get(1, TimeUnit.SECONDS);
          assertThat(result).isNotNull();
          assertThat(result.status()).isEqualTo("OK");
          assertThat(result.isOk()).isTrue();
  
          validateResponseServed(ApiEndpointsV1.CASSANDRA_JMX_HEALTH_ROUTE);
      }
  
      @Test
      void testCassandraJmxHealthNotOk() throws Exception
      {
          MockResponse response = new MockResponse()
                                  .setResponseCode(503)
                                  .setHeader("content-type", "application/json")
                                  .setBody("{\"status\":\"NOT_OK\"}");
          enqueue(response);
  
          assertThatThrownBy(() -> client.cassandraJmxHealth().get(1, TimeUnit.SECONDS))
          .isInstanceOf(ExecutionException.class)
          .hasCauseInstanceOf(RetriesExhaustedException.class);
  
          validateResponseServed(ApiEndpointsV1.CASSANDRA_JMX_HEALTH_ROUTE);
      }
  
      @Test
      void testFullSchema() throws Exception
      {
          String fullSchemaAsString = "{\"schema\":\"CREATE KEYSPACE sample_ks.sample_table ...\"}";
          MockResponse response = new MockResponse().setResponseCode(OK.code()).setBody(fullSchemaAsString);
          enqueue(response);
  
          SchemaResponse result = client.fullSchema().get(30, TimeUnit.SECONDS);
          assertThat(result).isNotNull();
          assertThat(result.keyspace()).isNull();
          assertThat(result.schema()).isEqualTo("CREATE KEYSPACE sample_ks.sample_table ...");
  
          validateResponseServed(ApiEndpointsV1.ALL_KEYSPACES_SCHEMA_ROUTE);
      }
  
      @Test
      void testSchema() throws Exception
      {
          String schemaAsString = "{\"keyspace\":\"cycling\",\"schema\":\"CREATE KEYSPACE sample_ks.sample_table ...\"}";
          MockResponse response = new MockResponse().setResponseCode(OK.code()).setBody(schemaAsString);
          enqueue(response);
  
          SchemaResponse result = client.schema("cycling").get(30, TimeUnit.SECONDS);
          assertThat(result).isNotNull();
          assertThat(result.keyspace()).isEqualTo("cycling");
          assertThat(result.schema()).isEqualTo("CREATE KEYSPACE sample_ks.sample_table ...");
  
          validateResponseServed(ApiEndpointsV1.KEYSPACE_SCHEMA_ROUTE.replaceAll(KEYSPACE_PATH_PARAM,
                                                                                 "cycling"));
      }
  
      @Test
      void testRing() throws Exception
      {
          String schemaAsString = "[{\"datacenter\":\"dc\",\"address\":\"127.0.0.1\",\"port\":80,\"rack\":\"r1\"," +
                                  "\"status\":\"up\",\"state\":\"normal\",\"load\":\"1 KiB\",\"owns\":\"1%\"," +
                                  "\"token\":\"100\",\"fqdn\":\"local\",\"hostId\":\"000\"}]";
          MockResponse response = new MockResponse().setResponseCode(OK.code()).setBody(schemaAsString);
          enqueue(response);
  
          RingResponse result = client.ring("cycling").get(30, TimeUnit.SECONDS);
          assertThat(result).isNotNull()
                            .hasSize(1);
          RingEntry entry = result.iterator().next();
          assertThat(entry.datacenter()).isEqualTo("dc");
          assertThat(entry.address()).isEqualTo("127.0.0.1");
          assertThat(entry.port()).isEqualTo(80);
          assertThat(entry.rack()).isEqualTo("r1");
          assertThat(entry.status()).isEqualTo("up");
          assertThat(entry.state()).isEqualTo("normal");
          assertThat(entry.load()).isEqualTo("1 KiB");
          assertThat(entry.owns()).isEqualTo("1%");
          assertThat(entry.token()).isEqualTo("100");
          assertThat(entry.fqdn()).isEqualTo("local");
          assertThat(entry.hostId()).isEqualTo("000");
  
          validateResponseServed(ApiEndpointsV1.RING_WITH_KEYSPACE_ROUTE.replaceAll(KEYSPACE_PATH_PARAM,
                                                                                    "cycling"));
      }
  
      @Test
      public void testNodeSettings() throws Exception
      {
          String nodeSettingsAsString = "{\"partitioner\":\"test-partitioner\", \"releaseVersion\": \"4.0.0\"}";
          MockResponse response = new MockResponse().setResponseCode(OK.code()).setBody(nodeSettingsAsString);
          enqueue(response);
  
          NodeSettings result = client.nodeSettings().get(30, TimeUnit.SECONDS);
          assertThat(result).isNotNull();
          assertThat(result.partitioner()).isEqualTo("test-partitioner");
          assertThat(result.releaseVersion()).isEqualTo("4.0.0");
  
          validateResponseServed(ApiEndpointsV1.NODE_SETTINGS_ROUTE);
      }
  
      @Test
      public void testNodeSettingsFromSpecifiedInstance() throws Exception
      {
          String nodeSettingsAsString = "{\"partitioner\":\"test-partitioner\", \"releaseVersion\": \"4.0.0\"}";
          MockResponse response = new MockResponse().setResponseCode(OK.code()).setBody(nodeSettingsAsString);
          MockWebServer mockWebServer = servers.get(1);
          mockWebServer.enqueue(response);
  
          SidecarInstanceImpl instance = new SidecarInstanceImpl(mockWebServer.getHostName(), mockWebServer.getPort());
          NodeSettings result = client.nodeSettings(instance).get(30, TimeUnit.SECONDS);
          assertThat(result).isNotNull();
          assertThat(result.partitioner()).isEqualTo("test-partitioner");
          assertThat(result.releaseVersion()).isEqualTo("4.0.0");
  
          assertThat(mockWebServer.getRequestCount()).isEqualTo(1);
          RecordedRequest request = mockWebServer.takeRequest();
          assertThat(request.getPath()).isEqualTo(ApiEndpointsV1.NODE_SETTINGS_ROUTE);
      }
  
      @Test
      public void testGossipInfo() throws Exception
      {
          String gossipInfoAsString = "{\"/127.0.0.1:7000\":{\"generation\":\"1\",\"schema\":\"4994b214\"," +
                                      "\"rack\":\"r2\",\"heartbeat\":\"214\",\"releaseVersion\":\"4.0.7\"," +
                                      "\"sstableVersions\":\"big-nb\"}}";
          MockResponse response = new MockResponse().setResponseCode(OK.code()).setBody(gossipInfoAsString);
          enqueue(response);
  
          GossipInfoResponse result = client.gossipInfo().get(30, TimeUnit.SECONDS);
          assertThat(result).isNotNull().hasSize(1);
          String key = result.entrySet().iterator().next().getKey();
          GossipInfoResponse.GossipInfo gossipInfo = result.get(key);
          assertThat(gossipInfo.generation()).isEqualTo("1");
          assertThat(gossipInfo.schema()).isEqualTo("4994b214");
          assertThat(gossipInfo.rack()).isEqualTo("r2");
          assertThat(gossipInfo.heartbeat()).isEqualTo("214");
          assertThat(gossipInfo.releaseVersion()).isEqualTo("4.0.7");
          assertThat(gossipInfo.sstableVersions()).isEqualTo(Collections.singletonList("big-nb"));
  
          validateResponseServed(ApiEndpointsV1.GOSSIP_ROUTE);
      }
  
      @Test
      public void testGossipHealthOK() throws InterruptedException, ExecutionException, TimeoutException
      {
          SidecarInstanceImpl sidecarInstance = instances.get(3);
          String gossipHealthAsString = "{\"status\":\"OK\"}";
          MockResponse response = new MockResponse().setResponseCode(OK.code()).setBody(gossipHealthAsString);
          enqueue(response);
  
          HealthResponse result = client.gossipHealth(sidecarInstance).get(30, TimeUnit.SECONDS);
          assertThat(result).isNotNull();
          assertThat(result.status()).isEqualTo("OK");
  
          validateResponseServed(ApiEndpointsV1.GOSSIP_HEALTH_ROUTE);
      }
  
      @Test
      public void testGossipHealthNotOK() throws InterruptedException, ExecutionException, TimeoutException
      {
          SidecarInstanceImpl sidecarInstance = instances.get(3);
          String gossipHealthAsString = "{\"status\":\"NOT_OK\"}";
          MockResponse response = new MockResponse().setResponseCode(OK.code()).setBody(gossipHealthAsString);
          enqueue(response);
  
          HealthResponse result = client.gossipHealth(sidecarInstance).get(30, TimeUnit.SECONDS);
          assertThat(result).isNotNull();
          assertThat(result.status()).isEqualTo("NOT_OK");
  
          validateResponseServed(ApiEndpointsV1.GOSSIP_HEALTH_ROUTE);
      }
  
      @Test
      public void testTimeSkew() throws Exception
      {
          String timeSkewAsString = "{\"currentTime\":\"123456789\",\"allowableSkewInMinutes\":\"122\"}}";
          MockResponse response = new MockResponse().setResponseCode(OK.code()).setBody(timeSkewAsString);
          enqueue(response);
  
          TimeSkewResponse result = client.timeSkew().get(30, TimeUnit.SECONDS);
          assertThat(result).isNotNull();
          assertThat(result.allowableSkewInMinutes).isEqualTo(122);
          assertThat(result.currentTime).isEqualTo(123456789);
  
          validateResponseServed(ApiEndpointsV1.TIME_SKEW_ROUTE);
      }
  
      @Test
      public void testTimeSkewFromReplicaSet() throws Exception
      {
          String timeSkewAsString = "{\"currentTime\":\"5555555\",\"allowableSkewInMinutes\":\"24\"}}";
          MockResponse response = new MockResponse().setResponseCode(OK.code()).setBody(timeSkewAsString);
          enqueue(response);
  
          TimeSkewResponse result = client.timeSkew(instances.subList(1, 2)).get(30, TimeUnit.SECONDS);
          assertThat(result).isNotNull();
          assertThat(result.allowableSkewInMinutes).isEqualTo(24);
          assertThat(result.currentTime).isEqualTo(5555555);
  
          validateResponseServed(ApiEndpointsV1.TIME_SKEW_ROUTE);
      }
  
      @Test
      public void testTokenRangeReplicasFromReplicaSet() throws Exception
      {
          String keyspace = "test";
          String nodeAddress = "127.0.0.1";
          int port = 7000;
          String nodeWithPort = nodeAddress + ":" + port;
          String expectedRangeStart = "-9223372036854775808";
          String expectedRangeEnd = "9223372036854775807";
          String tokenRangeReplicasAsString = "{\"replicaMetadata\":{\"127.0.0.1:7000\":{" +
                                              "\"state\":\"Normal\"," +
                                              "\"status\":\"Up\"," +
                                              "\"fqdn\":\"localhost\"," +
                                              "\"address\":\"127.0.0.1\"," +
                                              "\"port\":7000," +
                                              "\"datacenter\":\"datacenter1\"}}," +
                                              "\"writeReplicas\":[{\"start\":\"-9223372036854775808\"," +
                                              "\"end\":\"9223372036854775807\",\"replicasByDatacenter\":" +
                                              "{\"datacenter1\":[\"127.0.0.1:7000\"]}}],\"readReplicas\":" +
                                              "[{\"start\":\"-9223372036854775808\",\"end\":\"9223372036854775807\"," +
                                              "\"replicasByDatacenter\":{\"datacenter1\":[\"127.0.0.1:7000\"]}}]}";
          MockResponse response = new MockResponse().setResponseCode(OK.code()).setBody(tokenRangeReplicasAsString);
          enqueue(response);
  
          TokenRangeReplicasResponse result = client.tokenRangeReplicas(instances.subList(1, 2), keyspace)
                                                    .get(30, TimeUnit.SECONDS);
          assertThat(result).isNotNull();
          assertThat(result.writeReplicas()).hasSize(1);
          TokenRangeReplicasResponse.ReplicaInfo writeReplica = result.writeReplicas().get(0);
          assertThat(writeReplica.start()).isEqualTo(expectedRangeStart);
          assertThat(writeReplica.end()).isEqualTo(expectedRangeEnd);
          assertThat(writeReplica.replicasByDatacenter()).containsKey("datacenter1");
          assertThat(writeReplica.replicasByDatacenter().get("datacenter1")).containsExactly(nodeWithPort);
          assertThat(result.readReplicas()).hasSize(1);
          TokenRangeReplicasResponse.ReplicaInfo readReplica = result.readReplicas().get(0);
          assertThat(readReplica.start()).isEqualTo(expectedRangeStart);
          assertThat(readReplica.end()).isEqualTo(expectedRangeEnd);
          assertThat(readReplica.replicasByDatacenter()).containsKey("datacenter1");
          assertThat(readReplica.replicasByDatacenter().get("datacenter1")).containsExactly(nodeWithPort);
          assertThat(result.replicaMetadata()).hasSize(1);
          TokenRangeReplicasResponse.ReplicaMetadata instanceMetadata = result.replicaMetadata().get(nodeWithPort);
          assertThat(instanceMetadata.state()).isEqualTo("Normal");
          assertThat(instanceMetadata.status()).isEqualTo("Up");
          assertThat(instanceMetadata.fqdn()).isEqualTo("localhost");
          assertThat(instanceMetadata.datacenter()).isEqualTo("datacenter1");
  
          validateResponseServed(ApiEndpointsV1.KEYSPACE_TOKEN_MAPPING_ROUTE.replaceAll(
          KEYSPACE_PATH_PARAM, keyspace));
      }
  
      @Test
      public void testListSnapshotFiles() throws Exception
      {
          String responseAsString = "{\"snapshotFilesInfo\":[{\"size\":15,\"host\":\"localhost1\",\"port\":2020," +
                                    "\"dataDirIndex\":1,\"snapshotName\":\"2023.04.11\",\"keySpaceName\":\"cycling\"," +
                                    "\"tableName\":\"cyclist_name\",\"fileName\":\"nb-203-big-TOC.txt\"}]}";
          MockResponse response = new MockResponse().setResponseCode(OK.code()).setBody(responseAsString);
          SidecarInstanceImpl sidecarInstance = instances.get(2);
          MockWebServer mockWebServer = servers.get(2);
          mockWebServer.enqueue(response);
  
          ListSnapshotFilesResponse result = client.listSnapshotFiles(sidecarInstance,
                                                                      "cycling",
                                                                      "cyclist_name",
                                                                      "2023.04.11")
                                                   .get(30, TimeUnit.SECONDS);
          assertThat(result).isNotNull();
          assertThat(result.snapshotFilesInfo()).hasSize(1);
          ListSnapshotFilesResponse.FileInfo fileInfo = result.snapshotFilesInfo().get(0);
          assertThat(fileInfo.size).isEqualTo(15);
          assertThat(fileInfo.host).isEqualTo("localhost1");
          assertThat(fileInfo.port).isEqualTo(2020);
          assertThat(fileInfo.dataDirIndex).isEqualTo(1);
          assertThat(fileInfo.snapshotName).isEqualTo("2023.04.11");
          assertThat(fileInfo.keySpaceName).isEqualTo("cycling");
          assertThat(fileInfo.tableName).isEqualTo("cyclist_name");
          assertThat(fileInfo.fileName).isEqualTo("nb-203-big-TOC.txt");
  
          assertThat(mockWebServer.getRequestCount()).isEqualTo(1);
          RecordedRequest request = mockWebServer.takeRequest();
          assertThat(request.getPath()).isEqualTo(ApiEndpointsV1.SNAPSHOTS_ROUTE
                                                  .replaceAll(KEYSPACE_PATH_PARAM, "cycling")
                                                  .replaceAll(ApiEndpointsV1.TABLE_PATH_PARAM, "cyclist_name")
                                                  .replaceAll(ApiEndpointsV1.SNAPSHOT_PATH_PARAM, "2023.04.11")
                                                  + "?includeSecondaryIndexFiles=true");
      }
  
      /**
       * CASSANDRASC-94 introduced a new field ({@code tableId}) to the payload when listing snapshots. We
       * need to make sure the client is able to handle the payload with the additional field (and ignore it).
       *
       * @throws Exception when the test fails
       */
      @Test
      public void testListSnapshotFilesPayloadWithTableId() throws Exception
      {
          String responseAsString = "{\"snapshotFilesInfo\":[{\"size\":15,\"host\":\"localhost1\",\"port\":2020," +
                                    "\"dataDirIndex\":1,\"snapshotName\":\"2023.04.11\",\"keySpaceName\":\"cycling\"," +
                                    "\"tableName\":\"cyclist_name\",\"tableId\":\"1234\",\"fileName\":" +
                                    "\"nb-203-big-TOC.txt\"}]}";
          MockResponse response = new MockResponse().setResponseCode(OK.code()).setBody(responseAsString);
          SidecarInstanceImpl sidecarInstance = instances.get(2);
          MockWebServer mockWebServer = servers.get(2);
          mockWebServer.enqueue(response);
  
          ListSnapshotFilesResponse result = client.listSnapshotFiles(sidecarInstance,
                                                                      "cycling",
                                                                      "cyclist_name",
                                                                      "2023.04.11")
                                                   .get(30, TimeUnit.SECONDS);
          assertThat(result).isNotNull();
          assertThat(result.snapshotFilesInfo()).hasSize(1);
          ListSnapshotFilesResponse.FileInfo fileInfo = result.snapshotFilesInfo().get(0);
          assertThat(fileInfo.size).isEqualTo(15);
          assertThat(fileInfo.host).isEqualTo("localhost1");
          assertThat(fileInfo.port).isEqualTo(2020);
          assertThat(fileInfo.dataDirIndex).isEqualTo(1);
          assertThat(fileInfo.snapshotName).isEqualTo("2023.04.11");
          assertThat(fileInfo.keySpaceName).isEqualTo("cycling");
          assertThat(fileInfo.tableName).isEqualTo("cyclist_name");
          assertThat(fileInfo.fileName).isEqualTo("nb-203-big-TOC.txt");
  
          assertThat(mockWebServer.getRequestCount()).isEqualTo(1);
          RecordedRequest request = mockWebServer.takeRequest();
          assertThat(request.getPath()).isEqualTo(ApiEndpointsV1.SNAPSHOTS_ROUTE
                                                  .replaceAll(KEYSPACE_PATH_PARAM, "cycling")
                                                  .replaceAll(ApiEndpointsV1.TABLE_PATH_PARAM, "cyclist_name")
                                                  .replaceAll(ApiEndpointsV1.SNAPSHOT_PATH_PARAM, "2023.04.11")
                                                  + "?includeSecondaryIndexFiles=true");
      }
  
      @Test
      public void testListSnapshotFilesWithoutSecondaryIndexFiles() throws Exception
      {
          String responseAsString = "{\"snapshotFilesInfo\":[{\"size\":15,\"host\":\"localhost1\",\"port\":2020," +
                                    "\"dataDirIndex\":1,\"snapshotName\":\"2023.04.11\",\"keySpaceName\":\"cycling\"," +
                                    "\"tableName\":\"cyclist_name\",\"fileName\":\"nb-203-big-TOC.txt\"}]}";
          MockResponse response = new MockResponse().setResponseCode(OK.code()).setBody(responseAsString);
          SidecarInstanceImpl sidecarInstance = instances.get(2);
          MockWebServer mockWebServer = servers.get(2);
          mockWebServer.enqueue(response);
  
          ListSnapshotFilesResponse result = client.listSnapshotFiles(sidecarInstance,
                                                                      "cycling",
                                                                      "cyclist_name",
                                                                      "2023.04.11",
                                                                      false)
                                                   .get(30, TimeUnit.SECONDS);
          assertThat(result).isNotNull();
          assertThat(result.snapshotFilesInfo()).hasSize(1);
          ListSnapshotFilesResponse.FileInfo fileInfo = result.snapshotFilesInfo().get(0);
          assertThat(fileInfo.size).isEqualTo(15);
          assertThat(fileInfo.host).isEqualTo("localhost1");
          assertThat(fileInfo.port).isEqualTo(2020);
          assertThat(fileInfo.dataDirIndex).isEqualTo(1);
          assertThat(fileInfo.snapshotName).isEqualTo("2023.04.11");
          assertThat(fileInfo.keySpaceName).isEqualTo("cycling");
          assertThat(fileInfo.tableName).isEqualTo("cyclist_name");
          assertThat(fileInfo.fileName).isEqualTo("nb-203-big-TOC.txt");
  
          assertThat(mockWebServer.getRequestCount()).isEqualTo(1);
          RecordedRequest request = mockWebServer.takeRequest();
          assertThat(request.getPath()).isEqualTo(ApiEndpointsV1.SNAPSHOTS_ROUTE
                                                  .replaceAll(KEYSPACE_PATH_PARAM, "cycling")
                                                  .replaceAll(ApiEndpointsV1.TABLE_PATH_PARAM, "cyclist_name")
                                                  .replaceAll(ApiEndpointsV1.SNAPSHOT_PATH_PARAM, "2023.04.11"));
      }
  
  
      @Test
      void testClearSnapshot() throws Exception
      {
          MockResponse response = new MockResponse().setResponseCode(OK.code());
          SidecarInstanceImpl sidecarInstance = instances.get(2);
          MockWebServer mockWebServer = servers.get(2);
          mockWebServer.enqueue(response);
  
          client.clearSnapshot(sidecarInstance, "cycling", "cyclist_name", "2023.04.11")
                .get(30, TimeUnit.SECONDS);
  
          assertThat(mockWebServer.getRequestCount()).isEqualTo(1);
          RecordedRequest request = mockWebServer.takeRequest();
          assertThat(request.getPath()).isEqualTo(ApiEndpointsV1.SNAPSHOTS_ROUTE
                                                  .replaceAll(KEYSPACE_PATH_PARAM, "cycling")
                                                  .replaceAll(ApiEndpointsV1.TABLE_PATH_PARAM, "cyclist_name")
                                                  .replaceAll(ApiEndpointsV1.SNAPSHOT_PATH_PARAM, "2023.04.11"));
          assertThat(request.getMethod()).isEqualTo("DELETE");
      }
  
      @Test
      void testCreateSnapshot() throws Exception
      {
          MockResponse response = new MockResponse().setResponseCode(OK.code());
          SidecarInstanceImpl sidecarInstance = instances.get(3);
          MockWebServer mockWebServer = servers.get(3);
          mockWebServer.enqueue(response);
  
          client.createSnapshot(sidecarInstance, "cycling", "cyclist_name", "2023.04.11")
                .get(30, TimeUnit.SECONDS);
  
          assertThat(mockWebServer.getRequestCount()).isEqualTo(1);
          assertThat(mockWebServer.getRequestCount()).isEqualTo(1);
          RecordedRequest request = mockWebServer.takeRequest();
          assertThat(request.getPath()).isEqualTo(ApiEndpointsV1.SNAPSHOTS_ROUTE
                                                  .replaceAll(KEYSPACE_PATH_PARAM, "cycling")
                                                  .replaceAll(ApiEndpointsV1.TABLE_PATH_PARAM, "cyclist_name")
                                                  .replaceAll(ApiEndpointsV1.SNAPSHOT_PATH_PARAM, "2023.04.11"));
          assertThat(request.getMethod()).isEqualTo("PUT");
      }
  
      @Test
      void testCreateSnapshotWithTTL() throws Exception
      {
          MockResponse response = new MockResponse().setResponseCode(OK.code());
          SidecarInstanceImpl sidecarInstance = instances.get(3);
          MockWebServer mockWebServer = servers.get(3);
          mockWebServer.enqueue(response);
  
          client.createSnapshot(sidecarInstance, "cycling", "cyclist_name", "2023.04.11", "2d")
                .get(30, TimeUnit.SECONDS);
  
          assertThat(mockWebServer.getRequestCount()).isEqualTo(1);
          RecordedRequest request = mockWebServer.takeRequest();
          String expected = ApiEndpointsV1.SNAPSHOTS_ROUTE
                            .replaceAll(KEYSPACE_PATH_PARAM, "cycling")
                            .replaceAll(ApiEndpointsV1.TABLE_PATH_PARAM, "cyclist_name")
                            .replaceAll(ApiEndpointsV1.SNAPSHOT_PATH_PARAM, "2023.04.11") + "?ttl=2d";
          assertThat(request.getPath()).isEqualTo(expected);
          assertThat(request.getMethod()).isEqualTo("PUT");
      }
  
      @Test
      void testCleanUploadSession() throws Exception
      {
          MockResponse response = new MockResponse().setResponseCode(OK.code());
          SidecarInstanceImpl sidecarInstance = instances.get(0);
          MockWebServer mockWebServer = servers.get(0);
          mockWebServer.enqueue(response);
  
          client.cleanUploadSession(sidecarInstance, "00000")
                .get(30, TimeUnit.SECONDS);
  
          assertThat(mockWebServer.getRequestCount()).isEqualTo(1);
          RecordedRequest request = mockWebServer.takeRequest();
          assertThat(request.getPath()).isEqualTo(ApiEndpointsV1.SSTABLE_CLEANUP_ROUTE
                                                  .replaceAll(ApiEndpointsV1.UPLOAD_ID_PATH_PARAM, "00000"));
          assertThat(request.getMethod()).isEqualTo("DELETE");
      }
  
      @Test
      void testSSTableImport() throws Exception
      {
          String responseAsString = "{\"success\":true,\"uploadId\":\"0000-0000\",\"keyspace\":\"cycling\"," +
                                    "\"tableName\":\"cyclist_name\"}";
          MockResponse response = new MockResponse().setResponseCode(OK.code()).setBody(responseAsString);
          SidecarInstanceImpl sidecarInstance = instances.get(0);
          MockWebServer mockWebServer = servers.get(0);
          mockWebServer.enqueue(response);
  
          ImportSSTableRequest.ImportOptions options = new ImportSSTableRequest.ImportOptions();
          SSTableImportResponse result = client.importSSTableRequest(sidecarInstance,
                                                                     "cycling",
                                                                     "cyclist_name",
                                                                     "0000-0000",
                                                                     options)
                                               .get(30, TimeUnit.SECONDS);
  
          assertThat(result).isNotNull();
          assertThat(result.keyspace()).isEqualTo("cycling");
          assertThat(result.tableName()).isEqualTo("cyclist_name");
          assertThat(result.success()).isTrue();
          assertThat(result.uploadId()).isEqualTo("0000-0000");
  
          assertThat(mockWebServer.getRequestCount()).isEqualTo(1);
          RecordedRequest request = mockWebServer.takeRequest();
          assertThat(request.getPath()).isEqualTo(ApiEndpointsV1.SSTABLE_IMPORT_ROUTE
                                                  .replaceAll(KEYSPACE_PATH_PARAM, "cycling")
                                                  .replaceAll(ApiEndpointsV1.TABLE_PATH_PARAM, "cyclist_name")
                                                  .replaceAll(ApiEndpointsV1.UPLOAD_ID_PATH_PARAM, "0000-0000"));
          assertThat(request.getMethod()).isEqualTo("PUT");
      }
  
      @Test
      void testSSTableImportWithAcceptedResponse() throws Exception
      {
          String responseAsString = "{\"success\":true,\"uploadId\":\"0000-0000\",\"keyspace\":\"cycling\"," +
                                    "\"tableName\":\"cyclist_name\"}";
          SidecarInstanceImpl sidecarInstance = instances.get(0);
          MockWebServer mockWebServer = servers.get(0);
          mockWebServer.enqueue(new MockResponse().setResponseCode(ACCEPTED.code()));
          mockWebServer.enqueue(new MockResponse().setResponseCode(ACCEPTED.code()));
          mockWebServer.enqueue(new MockResponse().setResponseCode(ACCEPTED.code()));
          mockWebServer.enqueue(new MockResponse().setResponseCode(ACCEPTED.code()));
          mockWebServer.enqueue(new MockResponse().setResponseCode(OK.code()).setBody(responseAsString));
  
          ImportSSTableRequest.ImportOptions options = new ImportSSTableRequest.ImportOptions();
          SSTableImportResponse result = client.importSSTableRequest(sidecarInstance,
                                                                     "cycling",
                                                                     "cyclist_name",
                                                                     "0000-0000",
                                                                     options)
                                               .get(30, TimeUnit.SECONDS);
  
          assertThat(result).isNotNull();
          assertThat(result.keyspace()).isEqualTo("cycling");
          assertThat(result.tableName()).isEqualTo("cyclist_name");
          assertThat(result.success()).isTrue();
          assertThat(result.uploadId()).isEqualTo("0000-0000");
  
          assertThat(mockWebServer.getRequestCount()).isEqualTo(5);
          RecordedRequest request = mockWebServer.takeRequest();
          assertThat(request.getPath()).isEqualTo(ApiEndpointsV1.SSTABLE_IMPORT_ROUTE
                                                  .replaceAll(KEYSPACE_PATH_PARAM, "cycling")
                                                  .replaceAll(ApiEndpointsV1.TABLE_PATH_PARAM, "cyclist_name")
                                                  .replaceAll(ApiEndpointsV1.UPLOAD_ID_PATH_PARAM, "0000-0000"));
          assertThat(request.getMethod()).isEqualTo("PUT");
      }
  
      @Test
      void testUploadSSTableFailsWhenFileDoesNotExist()
      {
  
          assertThatIllegalArgumentException()
          .isThrownBy(() -> client.uploadSSTableRequest(new SidecarInstanceImpl("host", 8080),
                                                        "cycling",
                                                        "cyclist_name",
                                                        "0000-0000",
                                                        "nb-1-big-TOC.txt",
                                                        null,
                                                        "path")
                                  .get(30, TimeUnit.SECONDS))
          .withMessage("File 'path' does not exist");
      }
  
      @Test
      void testUploadSSTableWithoutDigest(@TempDir Path tempDirectory) throws Exception
      {
          Path fileToUpload = prepareFile(tempDirectory);
          try (MockWebServer server = new MockWebServer())
          {
              server.enqueue(new MockResponse().setResponseCode(OK.code()));
  
              SidecarInstanceImpl sidecarInstance = RequestExecutorTest.newSidecarInstance(server);
              client.uploadSSTableRequest(sidecarInstance,
                                          "cycling",
                                          "cyclist_name",
                                          "0000-0000",
                                          "nb-1-big-TOC.txt",
                                          null,
                                          fileToUpload.toString())
                    .get(30, TimeUnit.SECONDS);
  
              assertThat(server.getRequestCount()).isEqualTo(1);
              RecordedRequest request = server.takeRequest();
              assertThat(request.getPath())
              .isEqualTo(ApiEndpointsV1.SSTABLE_UPLOAD_ROUTE
                         .replaceAll(ApiEndpointsV1.UPLOAD_ID_PATH_PARAM, "0000-0000")
                         .replaceAll(KEYSPACE_PATH_PARAM, "cycling")
                         .replaceAll(ApiEndpointsV1.TABLE_PATH_PARAM, "cyclist_name")
                         .replaceAll(ApiEndpointsV1.COMPONENT_PATH_PARAM, "nb-1-big-TOC.txt"));
              assertThat(request.getMethod()).isEqualTo("PUT");
              assertThat(request.getHeader(HttpHeaderNames.CONTENT_MD5.toString())).isNull();
              assertThat(request.getHeader(HttpHeaderNames.CONTENT_LENGTH.toString())).isEqualTo("80");
              assertThat(request.getBodySize()).isEqualTo(80);
          }
      }
  
      @Test
      void testUploadSSTableWithMD5Digest(@TempDir Path tempDirectory) throws Exception
      {
          Path fileToUpload = prepareFile(tempDirectory);
          try (MockWebServer server = new MockWebServer())
          {
              server.enqueue(new MockResponse().setResponseCode(OK.code()));
  
              SidecarInstanceImpl sidecarInstance = RequestExecutorTest.newSidecarInstance(server);
              client.uploadSSTableRequest(sidecarInstance,
                                          "cycling",
                                          "cyclist_name",
                                          "0000-0000",
                                          "nb-1-big-TOC.txt",
                                          new MD5Digest("15a69dc6501aa5ae17af037fe053f610"),
                                          fileToUpload.toString())
                    .get(30, TimeUnit.SECONDS);
  
              assertThat(server.getRequestCount()).isEqualTo(1);
              RecordedRequest request = server.takeRequest();
              assertThat(request.getPath())
              .isEqualTo(ApiEndpointsV1.SSTABLE_UPLOAD_ROUTE
                         .replaceAll(ApiEndpointsV1.UPLOAD_ID_PATH_PARAM, "0000-0000")
                         .replaceAll(KEYSPACE_PATH_PARAM, "cycling")
                         .replaceAll(ApiEndpointsV1.TABLE_PATH_PARAM, "cyclist_name")
                         .replaceAll(ApiEndpointsV1.COMPONENT_PATH_PARAM, "nb-1-big-TOC.txt"));
              assertThat(request.getMethod()).isEqualTo("PUT");
              assertThat(request.getHeader(HttpHeaderNames.CONTENT_MD5.toString()))
              .isEqualTo("15a69dc6501aa5ae17af037fe053f610");
              assertThat(request.getHeader(HttpHeaderNames.CONTENT_LENGTH.toString())).isEqualTo("80");
              assertThat(request.getBodySize()).isEqualTo(80);
          }
      }
  
      @Test
      void testUploadSSTableWithXXHashDigest(@TempDir Path tempDirectory) throws Exception
      {
          Path fileToUpload = prepareFile(tempDirectory);
          try (MockWebServer server = new MockWebServer())
          {
              server.enqueue(new MockResponse().setResponseCode(OK.code()));
  
              SidecarInstanceImpl sidecarInstance = RequestExecutorTest.newSidecarInstance(server);
              client.uploadSSTableRequest(sidecarInstance,
                                          "cycling",
                                          "cyclist_name",
                                          "0000-0000",
                                          "nb-1-big-TOC.txt",
                                          new XXHash32Digest("15a69dc6501aa5ae17af037fe053f610"),
                                          fileToUpload.toString())
                    .get(30, TimeUnit.SECONDS);
  
              assertThat(server.getRequestCount()).isEqualTo(1);
              RecordedRequest request = server.takeRequest();
              assertThat(request.getPath())
              .isEqualTo(ApiEndpointsV1.SSTABLE_UPLOAD_ROUTE
                         .replaceAll(ApiEndpointsV1.UPLOAD_ID_PATH_PARAM, "0000-0000")
                         .replaceAll(KEYSPACE_PATH_PARAM, "cycling")
                         .replaceAll(ApiEndpointsV1.TABLE_PATH_PARAM, "cyclist_name")
                         .replaceAll(ApiEndpointsV1.COMPONENT_PATH_PARAM, "nb-1-big-TOC.txt"));
              assertThat(request.getMethod()).isEqualTo("PUT");
              assertThat(request.getHeader(CONTENT_XXHASH32))
              .isEqualTo("15a69dc6501aa5ae17af037fe053f610");
              assertThat(request.getHeader(CONTENT_XXHASH32_SEED)).isNull();
              assertThat(request.getHeader(HttpHeaderNames.CONTENT_LENGTH.toString())).isEqualTo("80");
              assertThat(request.getBodySize()).isEqualTo(80);
          }
      }
  
      @Test
      void testUploadSSTableWithXXHashDigestAndSeed(@TempDir Path tempDirectory) throws Exception
      {
          Path fileToUpload = prepareFile(tempDirectory);
          try (MockWebServer server = new MockWebServer())
          {
              server.enqueue(new MockResponse().setResponseCode(OK.code()));
  
              SidecarInstanceImpl sidecarInstance = RequestExecutorTest.newSidecarInstance(server);
              client.uploadSSTableRequest(sidecarInstance,
                                          "cycling",
                                          "cyclist_name",
                                          "0000-0000",
                                          "nb-1-big-TOC.txt",
                                          new XXHash32Digest("15a69dc6501aa5ae17af037fe053f610", "123456"),
                                          fileToUpload.toString())
                    .get(30, TimeUnit.SECONDS);
  
              assertThat(server.getRequestCount()).isEqualTo(1);
              RecordedRequest request = server.takeRequest();
              assertThat(request.getPath())
              .isEqualTo(ApiEndpointsV1.SSTABLE_UPLOAD_ROUTE
                         .replaceAll(ApiEndpointsV1.UPLOAD_ID_PATH_PARAM, "0000-0000")
                         .replaceAll(KEYSPACE_PATH_PARAM, "cycling")
                         .replaceAll(ApiEndpointsV1.TABLE_PATH_PARAM, "cyclist_name")
                         .replaceAll(ApiEndpointsV1.COMPONENT_PATH_PARAM, "nb-1-big-TOC.txt"));
              assertThat(request.getMethod()).isEqualTo("PUT");
              assertThat(request.getHeader(CONTENT_XXHASH32))
              .isEqualTo("15a69dc6501aa5ae17af037fe053f610");
              assertThat(request.getHeader(CONTENT_XXHASH32_SEED)).isEqualTo("123456");
              assertThat(request.getHeader(HttpHeaderNames.CONTENT_LENGTH.toString())).isEqualTo("80");
              assertThat(request.getBodySize()).isEqualTo(80);
          }
      }
  
      @ParameterizedTest
      @ValueSource(booleans = { true, false })
      void testLegacyStreamSSTableComponentWithNoRange(boolean useLegacyApi) throws Exception
      {
          try (MockWebServer server = new MockWebServer())
          {
              CountDownLatch latch = new CountDownLatch(1);
              List<byte[]> receivedBytes = new ArrayList<>();
              StreamConsumer mockStreamConsumer = new StreamConsumer()
              {
                  @Override
                  public void onRead(StreamBuffer buffer)
                  {
                      assertThat(buffer.readableBytes()).isGreaterThan(0);
                      byte[] dst = new byte[buffer.readableBytes()];
                      buffer.copyBytes(0, dst, 0, buffer.readableBytes());
                      receivedBytes.add(dst);
                  }
  
                  @Override
                  public void onComplete()
                  {
                      latch.countDown();
                  }
  
                  @Override
                  public void onError(Throwable throwable)
                  {
                      latch.countDown();
                      fail("Should not encounter an error", throwable);
                  }
              };
              InputStream inputStream = resourceInputStream("sstables/nb-1-big-TOC.txt");
              Buffer buffer = Okio.buffer(Okio.source(inputStream)).getBuffer();
              Okio.use(buffer, buffer1 -> {
                  try
                  {
                      return buffer1.writeAll(Okio.source(inputStream));
                  }
                  catch (IOException e)
                  {
                      throw new RuntimeException(e);
                  }
              });
  
              SidecarInstanceImpl sidecarInstance = RequestExecutorTest.newSidecarInstance(server);
              MockResponse response =
              new MockResponse().setResponseCode(OK.code())
                                .setHeader(HttpHeaderNames.CONTENT_TYPE.toString(),
                                           HttpHeaderValues.APPLICATION_OCTET_STREAM)
                                .setHeader(HttpHeaderNames.ACCEPT_RANGES.toString(), "bytes")
                                .setBody(buffer);
              server.enqueue(response);
  
              String expectedPath;
              if (useLegacyApi)
              {
                  client.streamSSTableComponent(sidecarInstance,
                                                "cycling",
                                                "cyclist_name",
                                                "2023.04.12",
                                                "nb-203-big-Data.db",
                                                null,
                                                mockStreamConsumer);
                  expectedPath = ApiEndpointsV1.COMPONENTS_ROUTE
                                 .replaceAll(KEYSPACE_PATH_PARAM, "cycling")
                                 .replaceAll(ApiEndpointsV1.TABLE_PATH_PARAM, "cyclist_name")
                                 .replaceAll(ApiEndpointsV1.SNAPSHOT_PATH_PARAM, "2023.04.12")
                                 .replaceAll(ApiEndpointsV1.COMPONENT_PATH_PARAM, "nb-203-big-Data.db");
              }
              else
              {
                  ListSnapshotFilesResponse.FileInfo fileInfo = new ListSnapshotFilesResponse.FileInfo(2023,
                                                                                                       server.getHostName(),
                                                                                                       server.getPort(), 0,
                                                                                                       "2023.04.12",
                                                                                                       "cycling",
                                                                                                       "cyclist_name-1234",
                                                                                                       "nb-1-big-TOC.txt");
                  client.streamSSTableComponent(sidecarInstance, fileInfo, null, mockStreamConsumer);
                  expectedPath = fileInfo.componentDownloadUrl();
              }
  
              assertThat(latch.await(1, TimeUnit.MINUTES)).isTrue();
  
              RecordedRequest request = server.takeRequest();
              assertThat(request.getPath()).isEqualTo(expectedPath);
              assertThat(request.getHeader("User-Agent")).isEqualTo("cassandra-sidecar-test/0.0.1");
              assertThat(request.getHeader("range")).isNull();
  
              byte[] bytes = receivedBytes.stream()
                                          .collect(ByteArrayOutputStream::new,
                                                   (outputStream, src) -> outputStream.write(src, 0, src.length),
                                                   (outputStream, src) -> {
                                                   })
                                          .toByteArray();
              assertThat(new String(bytes, StandardCharsets.UTF_8)).isEqualTo("Summary.db\n" +
                                                                              "TOC.txt\n" +
                                                                              "Statistics.db\n" +
                                                                              "Filter.db\n" +
                                                                              "Data.db\n" +
                                                                              "CRC.db\n" +
                                                                              "Digest.crc32\n" +
                                                                              "Index.db\n");
          }
      }
  
      @ParameterizedTest
      @ValueSource(booleans = { true, false })
      void testStreamSSTableComponentWithRange(boolean useLegacyApi) throws Exception
      {
          try (MockWebServer server = new MockWebServer())
          {
              CountDownLatch latch = new CountDownLatch(1);
              List<byte[]> receivedBytes = new ArrayList<>();
              StreamConsumer mockStreamConsumer = new StreamConsumer()
              {
                  @Override
                  public void onRead(StreamBuffer buffer)
                  {
                      assertThat(buffer.readableBytes()).isGreaterThan(0);
                      byte[] dst = new byte[buffer.readableBytes()];
                      buffer.copyBytes(0, dst, 0, buffer.readableBytes());
                      receivedBytes.add(dst);
                  }
  
                  @Override
                  public void onComplete()
                  {
                      latch.countDown();
                  }
  
                  @Override
                  public void onError(Throwable throwable)
                  {
                      latch.countDown();
                      fail("Should not encounter an error", throwable);
                  }
              };
  
              SidecarInstanceImpl sidecarInstance = RequestExecutorTest.newSidecarInstance(server);
              MockResponse response =
              new MockResponse().setResponseCode(PARTIAL_CONTENT.code())
                                .setHeader(HttpHeaderNames.CONTENT_TYPE.toString(),
                                           HttpHeaderValues.APPLICATION_OCTET_STREAM)
                                .setHeader(HttpHeaderNames.ACCEPT_RANGES.toString(), "bytes")
                                .setHeader(HttpHeaderNames.CONTENT_RANGE.toString(), "bytes 10-20/80")
                                .setBody("TOC.txt\nSt");
              server.enqueue(response);
  
              String expectedPath;
              if (useLegacyApi)
              {
                  client.streamSSTableComponent(sidecarInstance,
                                                "cycling",
                                                "cyclist_name",
                                                "2023.04.12",
                                                "nb-1-big-TOC.txt",
                                                HttpRange.of(10, 20),
                                                mockStreamConsumer);
                  expectedPath = ApiEndpointsV1.COMPONENTS_ROUTE
                                 .replaceAll(KEYSPACE_PATH_PARAM, "cycling")
                                 .replaceAll(ApiEndpointsV1.TABLE_PATH_PARAM, "cyclist_name")
                                 .replaceAll(ApiEndpointsV1.SNAPSHOT_PATH_PARAM, "2023.04.12")
                                 .replaceAll(ApiEndpointsV1.COMPONENT_PATH_PARAM, "nb-1-big-TOC.txt");
              }
              else
              {
                  ListSnapshotFilesResponse.FileInfo fileInfo = new ListSnapshotFilesResponse.FileInfo(2023,
                                                                                                       server.getHostName(),
                                                                                                       server.getPort(), 0,
                                                                                                       "2023.04.12",
                                                                                                       "cycling",
                                                                                                       "cyclist_name-1234",
                                                                                                       "nb-1-big-TOC.txt");
  
                  client.streamSSTableComponent(sidecarInstance, fileInfo, HttpRange.of(10, 20), mockStreamConsumer);
                  expectedPath = fileInfo.componentDownloadUrl();
              }
  
              assertThat(latch.await(1, TimeUnit.MINUTES)).isTrue();
  
              RecordedRequest request = server.takeRequest();
              assertThat(request.getPath()).isEqualTo(expectedPath);
              assertThat(request.getHeader("User-Agent")).isEqualTo("cassandra-sidecar-test/0.0.1");
              assertThat(request.getHeader("range")).isEqualTo("bytes=10-20");
  
              byte[] bytes = receivedBytes.stream()
                                          .collect(ByteArrayOutputStream::new,
                                                   (outputStream, src) -> outputStream.write(src, 0, src.length),
                                                   (outputStream, src) -> {
                                                   })
                                          .toByteArray();
              assertThat(new String(bytes, StandardCharsets.UTF_8)).isEqualTo("TOC.txt\nSt");
          }
      }
  
      @ParameterizedTest
      @ValueSource(booleans = { true, false })
      void testStreamSSTableComponentFailsMidStream(boolean useLegacyApi) throws Exception
      {
          try (MockWebServer server = new MockWebServer())
          {
              CountDownLatch latch = new CountDownLatch(1);
              List<byte[]> receivedBytes = new ArrayList<>();
              StreamConsumer mockStreamConsumer = new StreamConsumer()
              {
                  @Override
                  public void onRead(StreamBuffer buffer)
                  {
                      assertThat(buffer.readableBytes()).isGreaterThan(0);
                      byte[] dst = new byte[buffer.readableBytes()];
                      buffer.copyBytes(0, dst, 0, buffer.readableBytes());
                      receivedBytes.add(dst);
                  }
  
                  @Override
                  public void onComplete()
                  {
                  }
  
                  @Override
                  public void onError(Throwable throwable)
                  {
                      latch.countDown();
                      assertThat(throwable).isNotNull();
                  }
              };
              InputStream inputStream = resourceInputStream("sstables/nb-1-big-TOC.txt");
              Buffer buffer = Okio.buffer(Okio.source(inputStream)).getBuffer();
              Okio.use(buffer, buffer1 -> {
                  try
                  {
                      return buffer1.writeAll(Okio.source(inputStream));
                  }
                  catch (IOException e)
                  {
                      throw new RuntimeException(e);
                  }
              });
  
              SidecarInstanceImpl sidecarInstance = RequestExecutorTest.newSidecarInstance(server);
              MockResponse response =
              new MockResponse().setResponseCode(OK.code())
                                .setHeader(HttpHeaderNames.CONTENT_TYPE.toString(),
                                           HttpHeaderValues.APPLICATION_OCTET_STREAM)
                                .setHeader(HttpHeaderNames.ACCEPT_RANGES.toString(), "bytes")
                                .setBody(buffer)
                                .setSocketPolicy(SocketPolicy.DISCONNECT_DURING_RESPONSE_BODY);
              server.enqueue(response);
  
              String expectedPath;
              if (useLegacyApi)
              {
                  client.streamSSTableComponent(sidecarInstance,
                                                "cycling",
                                                "cyclist_name",
                                                "2023.04.12",
                                                "nb-1-big-TOC.txt",
                                                null,
                                                mockStreamConsumer);
                  expectedPath = ApiEndpointsV1.COMPONENTS_ROUTE
                                 .replaceAll(KEYSPACE_PATH_PARAM, "cycling")
                                 .replaceAll(ApiEndpointsV1.TABLE_PATH_PARAM, "cyclist_name")
                                 .replaceAll(ApiEndpointsV1.SNAPSHOT_PATH_PARAM, "2023.04.12")
                                 .replaceAll(ApiEndpointsV1.COMPONENT_PATH_PARAM, "nb-1-big-TOC.txt");
              }
              else
              {
  
                  ListSnapshotFilesResponse.FileInfo fileInfo = new ListSnapshotFilesResponse.FileInfo(2023,
                                                                                                       server.getHostName(),
                                                                                                       server.getPort(), 0,
                                                                                                       "2023.04.12",
                                                                                                       "cycling",
                                                                                                       "cyclist_name-1234",
                                                                                                       "nb-1-big-TOC.txt");
                  client.streamSSTableComponent(sidecarInstance, fileInfo, null, mockStreamConsumer);
                  expectedPath = fileInfo.componentDownloadUrl();
              }
  
              assertThat(latch.await(1, TimeUnit.MINUTES)).isTrue();
  
              RecordedRequest request = server.takeRequest();
              assertThat(request.getPath()).isEqualTo(expectedPath);
              assertThat(request.getHeader("User-Agent")).isEqualTo("cassandra-sidecar-test/0.0.1");
              assertThat(request.getHeader("range")).isNull();
              assertThat(receivedBytes).hasSizeGreaterThan(0);
          }
      }
  
      @ParameterizedTest
      @ValueSource(booleans = { true, false })
      void testStreamSSTableComponentWithRetries(boolean useLegacyApi) throws Exception
      {
          try (MockWebServer server = new MockWebServer())
          {
              CountDownLatch latch = new CountDownLatch(1);
              List<byte[]> receivedBytes = new ArrayList<>();
              StreamConsumer mockStreamConsumer = new StreamConsumer()
              {
                  @Override
                  public void onRead(StreamBuffer buffer)
                  {
                      assertThat(buffer.readableBytes()).isGreaterThan(0);
                      byte[] dst = new byte[buffer.readableBytes()];
                      buffer.copyBytes(0, dst, 0, buffer.readableBytes());
                      receivedBytes.add(dst);
                  }
  
                  @Override
                  public void onComplete()
                  {
                      latch.countDown();
                  }
  
                  @Override
                  public void onError(Throwable throwable)
                  {
                  }
              };
  
              SidecarInstanceImpl sidecarInstance = RequestExecutorTest.newSidecarInstance(server);
              MockResponse response =
              new MockResponse().setResponseCode(PARTIAL_CONTENT.code())
                                .setHeader(HttpHeaderNames.CONTENT_TYPE.toString(),
                                           HttpHeaderValues.APPLICATION_OCTET_STREAM)
                                .setHeader(HttpHeaderNames.ACCEPT_RANGES.toString(), "bytes")
                                .setHeader(HttpHeaderNames.CONTENT_RANGE.toString(), "bytes 10-20/80")
                                .setBody("TOC.txt\nSt");
              server.enqueue(new MockResponse().setResponseCode(INTERNAL_SERVER_ERROR.code())
                                               .setBody("{\"error\":\"some error\"}"));
              server.enqueue(new MockResponse().setResponseCode(INTERNAL_SERVER_ERROR.code())
                                               .setBody("{\"error\":\"some error\"}"));
              server.enqueue(response);
  
              String expectedPath;
              if (useLegacyApi)
              {
                  client.streamSSTableComponent(sidecarInstance,
                                                "cycling",
                                                "cyclist_name",
                                                "2023.04.12",
                                                "nb-1-big-TOC.txt",
                                                HttpRange.of(10, 20),
                                                mockStreamConsumer);
                  expectedPath = ApiEndpointsV1.COMPONENTS_ROUTE
                                 .replaceAll(KEYSPACE_PATH_PARAM, "cycling")
                                 .replaceAll(ApiEndpointsV1.TABLE_PATH_PARAM, "cyclist_name")
                                 .replaceAll(ApiEndpointsV1.SNAPSHOT_PATH_PARAM, "2023.04.12")
                                 .replaceAll(ApiEndpointsV1.COMPONENT_PATH_PARAM, "nb-1-big-TOC.txt");
              }
              else
              {
                  ListSnapshotFilesResponse.FileInfo fileInfo = new ListSnapshotFilesResponse.FileInfo(2023,
                                                                                                       server.getHostName(),
                                                                                                       server.getPort(), 0,
                                                                                                       "2023.04.12",
                                                                                                       "cycling",
                                                                                                       "cyclist_name-1234",
                                                                                                       "nb-1-big-TOC.txt");
                  client.streamSSTableComponent(sidecarInstance, fileInfo, HttpRange.of(10, 20), mockStreamConsumer);
                  expectedPath = fileInfo.componentDownloadUrl();
              }
              assertThat(latch.await(1, TimeUnit.MINUTES)).isTrue();
  
              server.takeRequest(); // first 500
              server.takeRequest(); // second 500
              RecordedRequest request3 = server.takeRequest();
              assertThat(request3.getPath()).isEqualTo(expectedPath);
              assertThat(request3.getHeader("User-Agent")).isEqualTo("cassandra-sidecar-test/0.0.1");
              assertThat(request3.getHeader("range")).isEqualTo("bytes=10-20");
  
              byte[] bytes = receivedBytes.stream()
                                          .collect(ByteArrayOutputStream::new,
                                                   (outputStream, src) -> outputStream.write(src, 0, src.length),
                                                   (outputStream, src) -> {
                                                   })
                                          .toByteArray();
              assertThat(new String(bytes, StandardCharsets.UTF_8)).isEqualTo("TOC.txt\nSt");
          }
      }
  
      @Test
      public void testOperationalJobs() throws Exception
      {
          UUID jobId = UUID.randomUUID();
          String jobStatusAsString = "{\"jobId\":\"" + jobId + "\",\"jobStatus\":\"RUNNING\",\"operation\":\"test\"}";
  
          MockResponse response = new MockResponse()
                                  .setResponseCode(OK.code())
                                  .setHeader("content-type", "application/json")
                                  .setBody(jobStatusAsString);
          enqueue(response);
  
          for (MockWebServer server : servers)
          {
              SidecarInstanceImpl sidecarInstance = RequestExecutorTest.newSidecarInstance(server);
              OperationalJobResponse result = client.operationalJobs(sidecarInstance, jobId).get(30, TimeUnit.SECONDS);
              assertThat(result).isNotNull();
              assertThat(result.jobId()).isEqualTo(jobId);
              assertThat(result.status()).isEqualTo(OperationalJobStatus.RUNNING);
              assertThat(result.operation()).isEqualTo("test");
              validateResponseServed(server,
                                     ApiEndpointsV1.OPERATIONAL_JOB_ROUTE.replaceAll(OPERATIONAL_JOB_ID_PATH_PARAM, jobId.toString()),
                                     req -> {});
          }
      }
  
      @Test
      public void testlistOperationalJobs() throws Exception
      {
          UUID jobId = UUID.randomUUID();
          String listJobsString = "{\"jobs\":[{\"jobId\":\"" + jobId + "\",\"status\":\"RUNNING\",\"failureReason\":\"\",\"operation\":\"test\"}]}";
  
          MockResponse response = new MockResponse()
                                  .setResponseCode(OK.code())
                                  .setHeader("content-type", "application/json")
                                  .setBody(listJobsString);
  
          enqueue(response);
  
          for (MockWebServer server : servers)
          {
              SidecarInstanceImpl sidecarInstance = RequestExecutorTest.newSidecarInstance(server);
              ListOperationalJobsResponse result = client.listOperationalJobs(sidecarInstance).get(30, TimeUnit.SECONDS);
              assertThat(result).isNotNull();
              assertThat(result.jobs()).isNotNull();
              assertThat(result.jobs().get(0).jobId()).isEqualTo(jobId);
              validateResponseServed(server, ApiEndpointsV1.LIST_OPERATIONAL_JOBS_ROUTE, req -> {});
          }
      }
  
      @Test
      public void testNodeDecommission() throws Exception
      {
          UUID jobId = UUID.randomUUID();
          String nodeDecommissionString = "{\"jobId\":\"" + jobId + "\",\"jobStatus\":\"SUCCEEDED\",\"instance\":\"127.0.0.1\"}";
  
          MockResponse response = new MockResponse()
                                  .setResponseCode(OK.code())
                                  .setHeader("content-type", "application/json")
                                  .setBody(nodeDecommissionString);
          enqueue(response);
  
          SidecarInstanceImpl sidecarInstance = RequestExecutorTest.newSidecarInstance(servers.get(0));
          OperationalJobResponse result = client.nodeDecommission(sidecarInstance).get(30, TimeUnit.SECONDS);
          assertThat(result).isNotNull();
          assertThat(result.status()).isEqualTo(OperationalJobStatus.SUCCEEDED);
          validateResponseServed(ApiEndpointsV1.NODE_DECOMMISSION_ROUTE);
      }
  
      @Test
      void testFailsWithOneAttemptPerServer()
      {
          for (MockWebServer server : servers)
          {
              MockResponse response = new MockResponse().setResponseCode(INTERNAL_SERVER_ERROR.code())
                                                        .setBody("{\"error\":\"some error\"}");
              server.enqueue(response);
          }
  
          assertThatExceptionOfType(ExecutionException.class)
          .isThrownBy(() -> client.schema("cycling").get(30, TimeUnit.SECONDS))
          .withRootCauseInstanceOf(RetriesExhaustedException.class)
          .withMessageContaining("Unable to complete request '/api/v1/keyspaces/cycling/schema' after 4 attempts");
      }
  
      @Test
      void testProvidingCustomRetryPolicy() throws ExecutionException, InterruptedException, TimeoutException
      {
          String nodeSettingsAsString = "{\"partitioner\":\"test-partitioner\", \"releaseVersion\": \"4.0.0\"}";
          MockResponse response = new MockResponse().setResponseCode(ACCEPTED.code()).setBody(nodeSettingsAsString);
          enqueue(response);
  
          RequestContext requestContext =
          client.requestBuilder()
                .request(new NodeSettingsRequest())
                .retryPolicy(new RetryPolicy()
                {
                    @Override
                    public void onResponse(CompletableFuture<HttpResponse> responseFuture,
                                           Request request,
                                           HttpResponse response,
                                           Throwable throwable,
                                           int attempts,
                                           boolean canRetryOnADifferentHost,
                                           RetryAction retryAction)
                    {
                        if (response != null && response.statusCode() == ACCEPTED.code())
                        {
                            responseFuture.complete(response);
                        }
                        else
                        {
                            client.defaultRetryPolicy().onResponse(responseFuture,
                                                                   request,
                                                                   response,
                                                                   throwable,
                                                                   attempts,
                                                                   canRetryOnADifferentHost,
                                                                   retryAction);
                        }
                    }
                })
                .build();
          NodeSettings result = client.<NodeSettings>executeRequestAsync(requestContext).get(30, TimeUnit.SECONDS);
          assertThat(result).isNotNull();
          assertThat(result.partitioner()).isEqualTo("test-partitioner");
          assertThat(result.releaseVersion()).isEqualTo("4.0.0");
  
          validateResponseServed(ApiEndpointsV1.NODE_SETTINGS_ROUTE);
      }
  
      @Test
      void testAcceptCreateRestoreJobRequest() throws Exception
      {
          String jobIdStr = "8e5799a4-d277-11ed-8d85-6916bb9b8056";
          enqueue(new MockResponse()
                  .setResponseCode(OK.code())
                  .setBody("{\"jobId\":\"" + jobIdStr + "\",\"status\":\"CREATED\"}"));
  
          UUID jobId = UUID.fromString(jobIdStr);
          long expireAt = System.currentTimeMillis() + 10000;
          RestoreJobSecrets secrets = RestoreJobSecretsGen.genRestoreJobSecrets();
          CreateRestoreJobRequestPayload requestPayload = CreateRestoreJobRequestPayload.builder(secrets, expireAt)
                                                                                        .jobId(jobId)
                                                                                        .build();
          CreateRestoreJobResponsePayload responsePayload = client.createRestoreJob("cycling",
                                                                                    "rank_by_year_and_name",
                                                                                    requestPayload)
                                                                  .join();
  
          assertThat(responsePayload).isNotNull();
          assertThat(responsePayload.jobId()).isEqualTo(jobId);
          assertThat(responsePayload.status()).isEqualTo("CREATED");
  
          ObjectMapper mapper = new ObjectMapper();
          String expectedReqBodyString = mapper.writeValueAsString(requestPayload);
          validateResponseServed(ApiEndpointsV1.CREATE_RESTORE_JOB_ROUTE
                                 .replaceAll(KEYSPACE_PATH_PARAM, "cycling")
                                 .replaceAll(TABLE_PATH_PARAM, "rank_by_year_and_name")
                                 .replaceAll(JOB_ID_PATH_PARAM, jobIdStr),
                                 recordedRequest -> {
                                     String reqBodyString = recordedRequest.getBody()
                                                                           .readString(Charset.defaultCharset());
                                     assertThat(reqBodyString).isEqualTo(expectedReqBodyString);
                                 });
      }
  
      @Test
      void testCreateRestoreJobShouldNotRetryOnDifferentHostWithBadRequest() throws Exception
      {
          String jobIdStr = "8e5799a4-d277-11ed-8d85-6916bb9b8056";
          enqueue(new MockResponse()
                  .setResponseCode(BAD_REQUEST.code())
                  .setBody("{\"status\":\"Fail\"," +
                           "\"message\":\"Error while decoding values, check your request body\"}"));
  
          UUID jobId = UUID.fromString(jobIdStr);
          long expireAt = System.currentTimeMillis() + 10000;
          RestoreJobSecrets secrets = RestoreJobSecretsGen.genRestoreJobSecrets();
          CreateRestoreJobRequestPayload requestPayload = CreateRestoreJobRequestPayload.builder(secrets, expireAt)
                                                                                        .jobId(jobId)
                                                                                        .build();
          assertThatException().isThrownBy(() -> client.createRestoreJob("badkeyspace",
                                                                         "bad_table",
                                                                         requestPayload)
                                                       .join())
                               .withCauseInstanceOf(RetriesExhaustedException.class)
                               .withMessageContaining("Unable to complete request '/api/v1/keyspaces/" +
                                                      "badkeyspace/tables/bad_table/restore-jobs' after 1 attempt");
  
          ObjectMapper mapper = new ObjectMapper();
          String expectedReqBodyString = mapper.writeValueAsString(requestPayload);
          validateResponseServed(ApiEndpointsV1.CREATE_RESTORE_JOB_ROUTE
                                 .replaceAll(KEYSPACE_PATH_PARAM, "badkeyspace")
                                 .replaceAll(TABLE_PATH_PARAM, "bad_table")
                                 .replaceAll(JOB_ID_PATH_PARAM, jobIdStr),
                                 recordedRequest -> {
                                     String reqBodyString = recordedRequest.getBody()
                                                                           .readString(Charset.defaultCharset());
                                     assertThat(reqBodyString).isEqualTo(expectedReqBodyString);
                                 });
      }
  
      @Test
      public void testConnectedClientStats() throws Exception
      {
          String connectedClientStatsResponseAsString = "{\"clientConnections\":[{\"address\":\"127.0.0.1\",\"port\":54628" +
                                                        ",\"sslEnabled\":false,\"sslCipherSuite\":\"TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256\"" +
                                                        ",\"sslProtocol\":\"TLSv1.2\",\"protocolVersion\":\"5\",\"username\":\"anonymous\"" +
                                                        ",\"requestCount\":39,\"driverName\":\"DataStax Java Driver\"" +
                                                        ",\"driverVersion\":\"3.11.3\",\"keyspaceName\":\"test\"" +
                                                        ",\"authenticationMode\":\"MutualTls\"" +
                                                        ",\"authenticationMetadata\":{\"identity\":\"spiffe://test.cassandra.apache.org/unitTest/mtls\"}" +
                                                        ",\"clientOptions\":{\"CQL_VERSION\":\"3.4.6\",\"DRIVER_NAME\":\"DataStax Python Driver\"" +
                                                        ",\"DRIVER_VERSION\":\"3.25.0\"}}],\"totalConnectedClients\":1" +
                                                        ",\"connectionsByUser\":{\"anonymous\":1}}";
  
          MockResponse response = new MockResponse().setResponseCode(OK.code()).setBody(connectedClientStatsResponseAsString);
          enqueue(response);
  
          for (MockWebServer server : servers)
          {
              SidecarInstanceImpl sidecarInstance = RequestExecutorTest.newSidecarInstance(server);
              ConnectedClientStatsResponse result = client.connectedClientStats(sidecarInstance).get();
  
              assertThat(result).isNotNull();
              assertThat(result.clientConnections()).isNotNull().hasSize(1);
              assertThat(result.totalConnectedClients()).isNotNull().isEqualTo(1);
              assertThat(result.connectionsByUser()).isNotNull().containsKey("anonymous");
              ClientConnectionEntry entry = result.clientConnections().iterator().next();
              assertThat(entry.address()).isEqualTo("127.0.0.1");
              assertThat(entry.port()).isEqualTo(54628);
              assertThat(entry.sslEnabled()).isEqualTo(false);
              assertThat(entry.sslCipherSuite()).isEqualTo("TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256");
              assertThat(entry.sslProtocol()).isEqualTo("TLSv1.2");
              assertThat(entry.protocolVersion()).isEqualTo("5");
              assertThat(entry.username()).isEqualTo("anonymous");
              assertThat(entry.requestCount()).isEqualTo(39);
              assertThat(entry.driverName()).isEqualTo("DataStax Java Driver");
              assertThat(entry.driverVersion()).isEqualTo("3.11.3");
              assertThat(entry.keyspaceName()).isEqualTo("test");
              assertThat(entry.authenticationMode()).isEqualTo("MutualTls");
              assertThat(entry.authenticationMetadata()).containsKey("identity");
              assertThat(entry.clientOptions()).containsKeys("CQL_VERSION", "DRIVER_NAME", "DRIVER_VERSION");
              validateResponseServed(server, ApiEndpointsV1.CONNECTED_CLIENT_STATS_ROUTE, req -> {});
          }
      }
  
      @Test
      public void testTableStats() throws Exception
      {
          String testKeyspace = "testKeyspace";
          String testTable = "testTable";
          int expectedSstables = 10;
          long expectedSize = 1024;
          long expectedTotalSize = 2048;
          long expectedSnapshotSize = 100;
  
          TableStatsResponse tableStatsResponse = new TableStatsResponse(testKeyspace,
                                                                         testTable,
                                                                         expectedSstables,
                                                                         expectedSize,
                                                                         expectedTotalSize,
                                                                         expectedSnapshotSize);
          ObjectMapper mapper = new ObjectMapper();
          MockResponse response = new MockResponse().setResponseCode(OK.code())
                                                    .setBody(mapper.writeValueAsString(tableStatsResponse));
          enqueue(response);
  
          for (MockWebServer server : servers)
          {
              SidecarInstanceImpl sidecarInstance = RequestExecutorTest.newSidecarInstance(server);
              TableStatsResponse result = client.tableStats(sidecarInstance, testKeyspace, testTable).get();
  
              assertThat(result).isNotNull();
              assertThat(result.sstableCount()).isEqualTo(expectedSstables);
              assertThat(result.diskSpaceUsedBytes()).isEqualTo(expectedSize);
              assertThat(result.totalDiskSpaceUsedBytes()).isEqualTo(expectedTotalSize);
              assertThat(result.snapshotsSizeBytes()).isEqualTo(expectedSnapshotSize);
              validateResponseServed(server,
                                     ApiEndpointsV1.TABLE_STATS_ROUTE
                                     .replaceAll(KEYSPACE_PATH_PARAM, testKeyspace)
                                     .replaceAll(TABLE_PATH_PARAM, testTable),
                                     req -> {});
          }
      }
  
      @Test
      public void testCompactionStats() throws Exception
      {
          String compactionStatsResponseAsString = "{\"concurrentCompactors\":2," +
                                                   "\"pendingTasks\":{}," +
                                                   "\"totalPendingTasks\":0," +
                                                   "\"completedCompactions\":42," +
                                                   "\"dataCompacted\":1048576," +
                                                   "\"abortedCompactions\":0," +
                                                   "\"reducedCompactions\":0," +
                                                   "\"sstablesDroppedFromCompaction\":5," +
                                                   "\"completedCompactionsRate\":{\"meanRate\":1.23,\"fifteenMinuteRate\":0.45}," +
                                                   "\"activeCompactions\":[]," +
                                                   "\"activeCompactionsCount\":0," +
                                                   "\"activeCompactionsRemainingTime\":0}";
  
          MockResponse response = new MockResponse().setResponseCode(OK.code()).setBody(compactionStatsResponseAsString);
          enqueue(response);
  
          for (MockWebServer server : servers)
          {
              SidecarInstanceImpl sidecarInstance = RequestExecutorTest.newSidecarInstance(server);
              CompactionStatsResponse result = client.compactionStats(sidecarInstance).get(30, TimeUnit.SECONDS);
  
              assertThat(result).isNotNull();
              assertThat(result.concurrentCompactors()).isEqualTo(2);
              assertThat(result.totalPendingTasks()).isEqualTo(0);
              assertThat(result.completedCompactions()).isEqualTo(42);
              assertThat(result.dataCompacted()).isEqualTo(1048576);
              assertThat(result.abortedCompactions()).isEqualTo(0);
              assertThat(result.reducedCompactions()).isEqualTo(0);
              assertThat(result.sstablesDroppedFromCompaction()).isEqualTo(5);
              assertThat(result.activeCompactions()).isEmpty();
              assertThat(result.activeCompactionsCount()).isEqualTo(0);
              assertThat(result.activeCompactionsRemainingTime()).isEqualTo(0);
              assertThat(result.completedCompactionsRate()).isNotNull();
              assertThat(result.completedCompactionsRate().meanRate()).isEqualTo(1.23);
              assertThat(result.completedCompactionsRate().fifteenMinuteRate()).isEqualTo(0.45);
              validateResponseServed(server, ApiEndpointsV1.COMPACTION_STATS_ROUTE, req -> {});
          }
      }
  
      @Test
      public void testCompactionStatsServerError() throws Exception
      {
          MockResponse response = new MockResponse()
                                  .setResponseCode(503)
                                  .setHeader("content-type", "application/json")
                                  .setBody("{\"error\":\"Internal server error\"}");
          
          // Enqueue enough error responses to handle retries
          MockWebServer server = servers.get(0);
          server.enqueue(response);
          server.enqueue(response);
          server.enqueue(response);
          server.enqueue(response);
  
          assertThatThrownBy(() -> client.compactionStats(instances.get(0)).get(30, TimeUnit.SECONDS))
          .isInstanceOf(ExecutionException.class)
          .hasCauseInstanceOf(RetriesExhaustedException.class);
  
          // Validate that the server received the expected number of retry attempts
          assertThat(server.getRequestCount()).isEqualTo(4);
          for (int i = 0; i < 4; i++)
          {
              RecordedRequest request = server.takeRequest();
              assertThat(request.getPath()).isEqualTo(ApiEndpointsV1.COMPACTION_STATS_ROUTE);
          }
      }
  
  
      @Test
      public void testListCdcSegments() throws ExecutionException, InterruptedException, JsonProcessingException
      {
          List<CdcSegmentInfo> segments = Arrays.asList(new CdcSegmentInfo("commit-log1", 100, 100, true, 1732148713725L),
                                                        new CdcSegmentInfo("commit-log2", 100, 10, false, 1732148713725L));
          ListCdcSegmentsResponse listSegmentsResponse = new ListCdcSegmentsResponse("localhost", 9043, segments);
          ObjectMapper mapper = new ObjectMapper();
  
          MockResponse response = new MockResponse();
          response.setResponseCode(200);
          response.setHeader("content-type", "application/json");
          response.setBody(mapper.writeValueAsString(listSegmentsResponse));
          enqueue(response);
  
          SidecarInstance instance = instances.get(0);
          ListCdcSegmentsResponse result = client.listCdcSegments(instance).get();
          assertThat(result).isNotNull();
          assertThat(result).isEqualTo(listSegmentsResponse);
          validateResponseServed(ApiEndpointsV1.LIST_CDC_SEGMENTS_ROUTE);
      }
  
      @Test
      public void testStreamCdcSegments() throws InterruptedException
      {
          MockResponse response = new MockResponse();
          // mock reading the first 12 bytes, i.e. "Test Content" from a large blob (1024).
          response.setResponseCode(200)
                  .setHeader(HttpHeaderNames.CONTENT_TYPE.toString(),
                             HttpHeaderValues.APPLICATION_OCTET_STREAM)
                  .setHeader(HttpHeaderNames.ACCEPT_RANGES.toString(), "bytes")
                  .setHeader(HttpHeaderNames.CONTENT_RANGE.toString(), "bytes 0-11/1024")
                  .setBody("Test Content");
          enqueue(response);
  
          SidecarInstance instance = instances.get(0);
          CountDownLatch latch = new CountDownLatch(1);
          List<byte[]> receivedBytes = new ArrayList<>();
          StreamConsumer mockStreamConsumer = new StreamConsumer()
          {
              @Override
              public void onRead(StreamBuffer buffer)
              {
                  assertThat(buffer.readableBytes()).isGreaterThan(0);
                  byte[] dst = new byte[buffer.readableBytes()];
                  buffer.copyBytes(0, dst, 0, buffer.readableBytes());
                  receivedBytes.add(dst);
              }
  
              @Override
              public void onComplete()
              {
                  latch.countDown();
              }
  
              @Override
              public void onError(Throwable throwable)
              {
                  latch.countDown();
              }
          };
          client.streamCdcSegments(instance, "testSegment", HttpRange.of(0, 11), mockStreamConsumer);
          latch.await();
          ByteArrayOutputStream baos = new ByteArrayOutputStream();
          for (byte[] bytes : receivedBytes)
          {
              baos.write(bytes, 0, bytes.length);
          }
          assertThat(new String(baos.toByteArray(), StandardCharsets.UTF_8)).isEqualTo("Test Content");
      }
  
      @Test
      public void testReportSchemaSuccess()
      {
          MockResponse response = new MockResponse().setResponseCode(OK.code())
                                                    .setBody("{\"status\":\"OK\"}");
  
          enqueue(response);
  
          SidecarInstance instance = instances.get(0);
  
          assertThatNoException().isThrownBy(() -> client.reportSchema(instance).get());
      }
  
      @Test
      public void testReportSchemaFailure()
      {
          MockResponse response = new MockResponse().setResponseCode(INTERNAL_SERVER_ERROR.code())
                                                    .setBody(INTERNAL_SERVER_ERROR.reasonPhrase());
  
          enqueue(response);
  
          SidecarInstance instance = instances.get(0);
  
          assertThatThrownBy(() -> client.reportSchema(instance).get()).isExactlyInstanceOf(ExecutionException.class)
                                                                       .hasCauseInstanceOf(RetriesExhaustedException.class)
                                                                       .hasMessageContaining(Integer.toString(INTERNAL_SERVER_ERROR.code()))
                                                                       .hasMessageContaining(INTERNAL_SERVER_ERROR.reasonPhrase());
      }
  
      @Test
      public void testAllServiceSuccessTests() throws IOException, ExecutionException, InterruptedException
      {
          List<AllServicesConfigPayload.Service> services = new ArrayList<>();
          Map<String, String> kafkaConfigs = new HashMap<>();
          kafkaConfigs.put("k1", "v1");
          kafkaConfigs.put("k2", "v2");
          Map<String, String> cdcConfigs = new HashMap<>();
          cdcConfigs.put("k1", "v1");
          cdcConfigs.put("k2", "v2");
          services.add(new AllServicesConfigPayload.Service("kafka", kafkaConfigs));
          services.add(new AllServicesConfigPayload.Service("cdc", cdcConfigs));
          AllServicesConfigPayload expectedResponse = new AllServicesConfigPayload(services);
  
          MockResponse response = new MockResponse();
          response.setResponseCode(200);
          response.setHeader("content-type", "application/json");
          ObjectMapper mapper = new ObjectMapper();
          response.setBody(mapper.writeValueAsString(expectedResponse));
          enqueue(response);
          assertThat(client.allServicesConfig().get()).isEqualTo(expectedResponse);
          validateResponseServed(ApiEndpointsV1.SERVICES_CONFIG_ROUTE,
                                 request -> assertThat(request.getMethod()).isEqualTo("GET"));
      }
  
      @Test
      public void testUpdateConfigSuccessTests() throws IOException, ExecutionException, InterruptedException
      {
          Map<String, String> payload = new HashMap<>();
          payload.put("testKey", "testValue");
          UpdateCdcServiceConfigPayload putResponse = new UpdateCdcServiceConfigPayload(payload);
          MockResponse response = new MockResponse();
          response.setResponseCode(200);
          response.setHeader("content-type", "application/json");
          ObjectMapper mapper = new ObjectMapper();
          response.setBody(mapper.writeValueAsString(putResponse));
          enqueue(response);
          assertThat(client.updateCdcServiceConfig(Service.CDC, payload).get()).isEqualTo(putResponse);
          validateResponseServed(ApiEndpointsV1.SERVICE_CONFIG_ROUTE.replaceAll(ApiEndpointsV1.SERVICE_PARAM, "cdc"),
                                 request -> assertThat(request.getMethod()).isEqualTo("PUT"));
      }
  
      @Test
      public void testDeleteConfigTests() throws ExecutionException, InterruptedException
      {
          MockResponse response = new MockResponse();
          response.setResponseCode(200);
          enqueue(response);
          client.deleteCdcServiceConfig(Service.CDC).get();
          validateResponseServed(ApiEndpointsV1.SERVICE_CONFIG_ROUTE.replaceAll(ApiEndpointsV1.SERVICE_PARAM, "cdc"),
                                 request -> assertThat(request.getMethod()).isEqualTo("DELETE"));
      }
  
      @Test
      public void testStreamsStats() throws Exception
      {
          StreamsProgressStats stats = new StreamsProgressStats(7, 7, 15088, 15088, 2, 2, 1024, 1024);
          StreamStatsResponse mockResp = new StreamStatsResponse("NORMAL", stats);
          ObjectMapper mapper = new ObjectMapper();
          String expectedResponse = mapper.writeValueAsString(mockResp);
          MockResponse response = new MockResponse()
                                  .setResponseCode(OK.code())
                                  .setBody(expectedResponse);
          enqueue(response);
          for (MockWebServer server : servers)
          {
              SidecarInstanceImpl sidecarInstance = RequestExecutorTest.newSidecarInstance(server);
              StreamStatsResponse result = client.streamsStats(sidecarInstance).get(30, TimeUnit.SECONDS);
              assertThat(mapper.writeValueAsString(result)).isEqualTo(expectedResponse);
              validateResponseServed(server, ApiEndpointsV1.STREAM_STATS_ROUTE, req -> {});
          }
      }
  
      @Test
      void testNodeUpdateGossip() throws Exception
      {
          MockResponse response = new MockResponse().setResponseCode(200).setBody("{\"status\":\"OK\"}");
          enqueue(response);
  
          SidecarInstanceImpl sidecarInstance = instances.get(0);
          HealthResponse result = client.nodeUpdateGossip(sidecarInstance, NodeCommandRequestPayload.State.STOP).get(30, TimeUnit.SECONDS);
          assertThat(result).isNotNull();
          assertThat(result.status()).isEqualToIgnoringCase("OK");
          assertThat(result.isOk()).isTrue();
  
          validateResponseServed(ApiEndpointsV1.GOSSIP_ROUTE, request -> {
              String requestBody = request.getBody().readUtf8();
              assertThat(requestBody).isEqualTo("{\"state\":\"stop\"}");
          });
      }
  
      @Test
      void testNodeUpdateNative() throws Exception
      {
          MockResponse response = new MockResponse().setResponseCode(200).setBody("{\"status\":\"OK\"}");
          enqueue(response);
  
          SidecarInstanceImpl sidecarInstance = instances.get(0);
          HealthResponse result = client.nodeUpdateNative(sidecarInstance, NodeCommandRequestPayload.State.STOP).get(30, TimeUnit.SECONDS);
          assertThat(result).isNotNull();
          assertThat(result.status()).isEqualToIgnoringCase("OK");
          assertThat(result.isOk()).isTrue();
  
          validateResponseServed(ApiEndpointsV1.CASSANDRA_NATIVE_ROUTE, request -> {
              String requestBody = request.getBody().readUtf8();
              assertThat(requestBody).isEqualTo("{\"state\":\"stop\"}");
          });
      }
  
      @Test
      void testLiveMigrationListInstanceFiles() throws ExecutionException, InterruptedException, JsonProcessingException
      {
          List<InstanceFileInfo> fileInfoList = Arrays.asList(new InstanceFileInfo("commit-log1", 100, FILE, System.currentTimeMillis()),
                                                              new InstanceFileInfo("commit-log2", 100, DIRECTORY, System.currentTimeMillis()));
          InstanceFilesListResponse filesListResponse = new InstanceFilesListResponse(fileInfoList);
          ObjectMapper mapper = new ObjectMapper();
  
          MockResponse response = new MockResponse();
          response.setResponseCode(200);
          response.setHeader("content-type", "application/json");
          response.setBody(mapper.writeValueAsString(filesListResponse));
          enqueue(response);
  
          SidecarInstance instance = instances.get(0);
          InstanceFilesListResponse result = client.liveMigrationListInstanceFilesAsync(instance).get();
          assertThat(result).isNotNull();
          assertThat(result).isEqualTo(filesListResponse);
          validateResponseServed(ApiEndpointsV1.LIVE_MIGRATION_FILES_ROUTE);
      }
  
      @Test
      void testLiveMigrationStreamFileAsync(@TempDir Path tempDirectory) throws ExecutionException, InterruptedException, IOException
      {
          String dummyFileContent = "Some dummy file content";
          MockResponse response = new MockResponse();
          response.setResponseCode(200);
          response.setHeader("content-type", "application/text");
          response.setBody(dummyFileContent);
          enqueue(response);
  
          Path filePath = tempDirectory.resolve("test_file.txt");
          SidecarInstance instance = instances.get(0);
          String fileUrl = LIVE_MIGRATION_FILES_ROUTE + "/data/0/test_file.text";
  
          client.liveMigrationStreamFileAsync(instance, fileUrl, filePath.toString()).get();
  
          assertThat(Files.exists(filePath)).isTrue();
          assertThat(new String(Files.readAllBytes(filePath), StandardCharsets.UTF_8)).isEqualTo(dummyFileContent);
          validateResponseServed(fileUrl);
      }
  
      @Test
      void testLiveMigrationStreamFileAsyncFileNotFound(@TempDir Path tempDirectory) throws ExecutionException, InterruptedException
      {
          MockResponse response = new MockResponse();
          response.setResponseCode(404);
          enqueue(response);
  
          Path filePath = tempDirectory.resolve("test_file.txt");
          SidecarInstance instance = instances.get(0);
          String fileUrl = LIVE_MIGRATION_FILES_ROUTE + "/data/0/test_file.text";
  
          client.liveMigrationStreamFileAsync(instance, fileUrl, filePath.toString()).get();
  
          assertThat(Files.exists(filePath)).isFalse();
          validateResponseServed(fileUrl);
      }
  
      @Test
      void testLiveMigrationStreamFileAsyncBadRequest(@TempDir Path tempDirectory) throws InterruptedException
      {
          MockResponse response = new MockResponse();
          response.setResponseCode(400);
          enqueue(response);
  
          Path filePath = tempDirectory.resolve("test_file.txt");
          SidecarInstance instance = instances.get(0);
          String fileUrl = LIVE_MIGRATION_FILES_ROUTE + "/data/0/test_file.text";
  
          CompletableFuture<Void> result = client.liveMigrationStreamFileAsync(instance, fileUrl, filePath.toString());
  
          assertThatExceptionOfType(ExecutionException.class).isThrownBy(result::get);
          assertThat(result.isCompletedExceptionally()).isTrue();
          assertThat(Files.exists(filePath)).isFalse();
          validateResponseServed(fileUrl);
      }
  
      @Test
      void testNodeLifecycleInfo() throws Exception
      {
          String lifecycleInfoResponse = "{\"current_state\":\"RUNNING\",\"desired_state\":\"RUNNING\",\"status\":\"CONVERGED\"," +
                                        "\"last_update\":\"Instance has started\"}";
          MockResponse response = new MockResponse().setResponseCode(OK.code()).setBody(lifecycleInfoResponse);
          enqueue(response);
  
          SidecarInstanceImpl sidecarInstance = instances.get(0);
          LifecycleInfoResponse result = client.nodeLifecycleInfo(sidecarInstance).get(30, TimeUnit.SECONDS);
          assertThat(result).isNotNull();
          assertThat(result.currentState()).isEqualTo(CassandraState.RUNNING);
          assertThat(result.desiredState()).isEqualTo(CassandraState.RUNNING);
          assertThat(result.status()).isEqualTo(OperationStatus.CONVERGED);
          assertThat(result.lastUpdate()).isEqualTo("Instance has started");
  
          validateResponseServed(ApiEndpointsV1.LIFECYCLE_ROUTE);
      }
  
      @Test
      void testNodeUpdateLifecycle() throws Exception
      {
          String lifecycleUpdateResponse = "{\"current_state\":\"RUNNING\",\"desired_state\":\"STOPPED\",\"status\":\"CONVERGING\"," +
                  "\"last_update\":\"Submitting stop task for instance\"}";
          MockResponse response = new MockResponse().setResponseCode(OK.code()).setBody(lifecycleUpdateResponse);
          enqueue(response);
  
          SidecarInstanceImpl sidecarInstance = instances.get(0);
          LifecycleInfoResponse result = client.nodeUpdateLifecycle(sidecarInstance, NodeCommandRequestPayload.State.STOP)
                  .get(30, TimeUnit.SECONDS);
  
          assertThat(result).isNotNull();
          assertThat(result.currentState()).isEqualTo(CassandraState.RUNNING);
          assertThat(result.desiredState()).isEqualTo(CassandraState.STOPPED);
          assertThat(result.status()).isEqualTo(OperationStatus.CONVERGING);
          assertThat(result.lastUpdate()).isEqualTo("Submitting stop task for instance");
  
          validateResponseServed(ApiEndpointsV1.LIFECYCLE_ROUTE, request -> {
              String requestBody = request.getBody().readUtf8();
              assertThat(request.getMethod()).isEqualTo("PUT");
              assertThat(requestBody).isEqualTo("{\"state\":\"stop\"}");
          });
      }
  
      private void enqueue(MockResponse response)
      {
          for (MockWebServer server : servers)
          {
              server.enqueue(response);
          }
      }
  
      private void validateResponseServed(String expectedEndpointPath) throws InterruptedException
      {
          validateResponseServed(expectedEndpointPath, req -> {
          });
      }
  
      private void validateResponseServed(String expectedEndpointPath,
                                          Consumer<RecordedRequest> serverReceivedRequestVerifier) throws InterruptedException
      {
          for (MockWebServer server : servers)
          {
              if (validateResponseServed(server, expectedEndpointPath, serverReceivedRequestVerifier))
              {
                  return;
              }
          }
          fail("The request was not served by any of the provided servers");
      }
  
      private boolean validateResponseServed(MockWebServer server, String expectedEndpointPath, Consumer<RecordedRequest> serverReceivedRequestVerifier)
      throws InterruptedException
      {
          if (server.getRequestCount() > 0)
          {
              assertThat(server.getRequestCount()).isEqualTo(1);
              RecordedRequest request = server.takeRequest();
              serverReceivedRequestVerifier.accept(request);
              assertThat(request.getPath()).isEqualTo(expectedEndpointPath);
              return true;
          }
          return false;
      }
  
      private InputStream resourceInputStream(String name)
      {
          InputStream inputStream = getClass().getClassLoader().getResourceAsStream(name);
          assertThat(inputStream).isNotNull();
          return inputStream;
      }
  
      private Path prepareFile(Path tempDirectory) throws IOException
      {
          Path fileToUpload = tempDirectory.resolve("nb-1-big-TOC.txt");
          try (InputStream inputStream = resourceInputStream("sstables/nb-1-big-TOC.txt"))
          {
              Files.copy(inputStream, fileToUpload, StandardCopyOption.REPLACE_EXISTING);
          }
          return fileToUpload;
      }
  }
  