CASSSIDECAR-343: Sidecar endpoint for flushing a node

adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/CassandraStorageOperations.java

@@ -323,4 +323,14 @@ public int getCompactionThroughputMbPerSec()
         return jmxClient.proxy(StorageJmxOperations.class, STORAGE_SERVICE_OBJ_NAME)
                         .getCompactionThroughputMbPerSec();
     }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public void flush(@NotNull String keyspace, @NotNull String... tableNames) throws IOException
+    {
+        jmxClient.proxy(StorageJmxOperations.class, STORAGE_SERVICE_OBJ_NAME)
+                 .forceKeyspaceFlush(keyspace, tableNames);
+    }
 }

adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/jmx/GossipDependentStorageJmxOperations.java

@@ -219,4 +219,10 @@ public int getCompactionThroughputMbPerSec()
     {
         return delegate.getCompactionThroughputMbPerSec();
     }
+
+    @Override
+    public void forceKeyspaceFlush(String keyspaceName, String... tableNames) throws IOException
+    {
+        delegate.forceKeyspaceFlush(keyspaceName, tableNames);
+    }
 }

adapters/adapters-base/src/main/java/org/apache/cassandra/sidecar/adapters/base/jmx/StorageJmxOperations.java

@@ -228,4 +228,13 @@ public interface StorageJmxOperations
      * @return the current compaction throughput in megabytes per second, or 0 if throughput cannot be determined
      */
     int getCompactionThroughputMbPerSec();
+
+    /**
+     * Triggers the node flush operation to flush memtables for the specified keyspace and tables.
+     *
+     * @param keyspaceName the keyspace name
+     * @param tableNames   the array of table names to flush; if empty, all tables in the keyspace will be flushed
+     * @throws IOException if an I/O error occurs during the flush operation
+     */
+    void forceKeyspaceFlush(String keyspaceName, String... tableNames) throws IOException;
 }

client-common/src/main/java/org/apache/cassandra/sidecar/common/ApiEndpointsV1.java

@@ -140,6 +140,7 @@ public final class ApiEndpointsV1
     public static final String LIST_OPERATIONAL_JOBS_ROUTE = API_V1 + CASSANDRA + OPERATIONAL_JOBS;
     public static final String OPERATIONAL_JOB_ROUTE = API_V1 + CASSANDRA + PER_OPERATIONAL_JOB;
     public static final String NODE_DECOMMISSION_ROUTE = API_V1 + CASSANDRA + "/operations/decommission";
+    public static final String NODE_FLUSH_ROUTE = API_V1 + CASSANDRA + PER_KEYSPACE + "/flush";
     public static final String STREAM_STATS_ROUTE = API_V1 + CASSANDRA + "/stats/streams";
     public static final String TABLE_STATS_ROUTE = API_V1 + CASSANDRA + PER_KEYSPACE + PER_TABLE + "/stats";
 

client-common/src/main/java/org/apache/cassandra/sidecar/common/request/NodeFlushRequest.java

@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.common.request;
+
+import java.util.List;
+
+import io.netty.handler.codec.http.HttpMethod;
+import org.apache.cassandra.sidecar.common.ApiEndpointsV1;
+import org.apache.cassandra.sidecar.common.request.data.NodeFlushRequestPayload;
+import org.apache.cassandra.sidecar.common.response.OperationalJobResponse;
+
+/**
+ * Represents a node flush request
+ */
+public class NodeFlushRequest extends JsonRequest<OperationalJobResponse>
+{
+    private final NodeFlushRequestPayload payload;
+
+    /**
+     * Constructs a NodeFlushRequest for the given keyspace and table names
+     *
+     * @param keyspace   the keyspace name
+     * @param tableNames the list of table names to flush (can be empty)
+     */
+    public NodeFlushRequest(String keyspace, List<String> tableNames)
+    {
+        super(ApiEndpointsV1.NODE_FLUSH_ROUTE.replace(ApiEndpointsV1.KEYSPACE_PATH_PARAM, keyspace));
+        this.payload = new NodeFlushRequestPayload(tableNames);
+    }
+
+    /**
+     * @return the list of table names to flush
+     */
+    public List<String> tableNames()
+    {
+        return payload.tableNames();
+    }
+
+    @Override
+    public HttpMethod method()
+    {
+        return HttpMethod.POST;
+    }
+
+    @Override
+    public Object requestBody()
+    {
+        return payload;
+    }
+}

client-common/src/main/java/org/apache/cassandra/sidecar/common/request/data/NodeFlushRequestPayload.java

@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.common.request.data;
+
+import java.util.Collections;
+import java.util.List;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Represents the request payload for node flush operation
+ */
+@JsonIgnoreProperties(ignoreUnknown = true)
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class NodeFlushRequestPayload
+{
+    private final List<String> tableNames;
+
+    /**
+     * Constructs a NodeFlushRequestPayload
+     *
+     * @param tableNames the list of table names to flush; null or empty list means flush all tables
+     */
+    @JsonCreator
+    public NodeFlushRequestPayload(@JsonProperty("tableNames") List<String> tableNames)
+    {
+        this.tableNames = tableNames != null ? tableNames : Collections.emptyList();
+    }
+
+    /**
+     * @return the list of table names to flush
+     */
+    @NotNull
+    @JsonProperty("tableNames")
+    public List<String> tableNames()
+    {
+        return tableNames;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public String toString()
+    {
+        return "NodeFlushRequestPayload{tableNames=" + tableNames + "}";
+    }
+}

client/src/main/java/org/apache/cassandra/sidecar/client/SidecarClient.java

@@ -19,6 +19,7 @@
 
 package org.apache.cassandra.sidecar.client;
 
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -45,6 +46,7 @@
 import org.apache.cassandra.sidecar.common.request.ImportSSTableRequest;
 import org.apache.cassandra.sidecar.common.request.ListCdcSegmentsRequest;
 import org.apache.cassandra.sidecar.common.request.LiveMigrationListInstanceFilesRequest;
+import org.apache.cassandra.sidecar.common.request.NodeFlushRequest;
 import org.apache.cassandra.sidecar.common.request.RestoreJobProgressRequest;
 import org.apache.cassandra.sidecar.common.request.RestoreJobSummaryRequest;
 import org.apache.cassandra.sidecar.common.request.Service;
@@ -823,6 +825,36 @@ public CompletableFuture<OperationalJobResponse> nodeDecommission(SidecarInstanc
                                             .build());
     }
 
+    /**
+     * Executes the node flush request using the default retry policy and provided {@code instance}.
+     * Flushes memtables for the specified keyspace and optionally for specific tables.
+     *
+     * @param instance   the instance where the request will be executed
+     * @param keyspace   the keyspace name to flush
+     * @param tableNames the list of table names to flush (can be empty to flush all tables)
+     * @return a completable future of the operational job response
+     */
+    public CompletableFuture<OperationalJobResponse> nodeFlush(SidecarInstance instance, String keyspace, List<String> tableNames)
+    {
+        return executor.executeRequestAsync(requestBuilder()
+                                            .singleInstanceSelectionPolicy(instance)
+                                            .request(new NodeFlushRequest(keyspace, tableNames))
+                                            .build());
+    }
+
+    /**
+     * Executes the node flush request using the default retry policy and provided {@code instance}.
+     * Flushes all memtables for the specified keyspace.
+     *
+     * @param instance the instance where the request will be executed
+     * @param keyspace the keyspace name to flush
+     * @return a completable future of the operational job response
+     */
+    public CompletableFuture<OperationalJobResponse> nodeFlush(SidecarInstance instance, String keyspace)
+    {
+        return nodeFlush(instance, keyspace, Collections.emptyList());
+    }
+
     /**
      * Sends a request to start or stop Cassandra gossiping on the provided instance.
      * <p>