diff --git a/test/nodetool/test_flush.py b/test/nodetool/test_flush.py
new file mode 100644
index 0000000000..779dc40375
--- /dev/null
+++ b/test/nodetool/test_flush.py
@@ -0,0 +1,38 @@
+#
+# Copyright 2023-present ScyllaDB
+#
+# SPDX-License-Identifier: AGPL-3.0-or-later
+#
+
+from rest_api_mock import expected_request
+import utils
+
+
+def test_flush(nodetool):
+    nodetool("flush", "ks1", expected_requests=[
+        expected_request("GET", "/storage_service/keyspaces", response=["ks1", "ks2"]),
+        expected_request("POST", "/storage_service/keyspace_flush/ks1")
+    ])
+
+
+def test_flush_one_table(nodetool):
+    nodetool("flush", "ks1", "tbl1", expected_requests=[
+        expected_request("GET", "/storage_service/keyspaces", response=["ks1", "ks2"]),
+        expected_request("POST", "/storage_service/keyspace_flush/ks1", params={"cf": "tbl1"})
+    ])
+
+
+def test_flush_two_tables(nodetool):
+    nodetool("flush", "ks1", "tbl1", "tbl2", expected_requests=[
+        expected_request("GET", "/storage_service/keyspaces", response=["ks1", "ks2"]),
+        expected_request("POST", "/storage_service/keyspace_flush/ks1", params={"cf": "tbl1,tbl2"})
+    ])
+
+
+def test_flush_none_existent_keyspace(nodetool):
+    utils.check_nodetool_fails_with(
+            nodetool,
+            ("flush", "non_existent_ks"),
+            {"expected_requests": [expected_request("GET", "/storage_service/keyspaces", response=["ks1", "ks2"])]},
+            ["nodetool: Keyspace [non_existent_ks] does not exist.",
+             "error processing arguments: keyspace non_existent_ks does not exist"])
diff --git a/tools/scylla-nodetool.cc b/tools/scylla-nodetool.cc
index d8832d2d22..d0267c7cae 100644
--- a/tools/scylla-nodetool.cc
+++ b/tools/scylla-nodetool.cc
@@ -123,6 +123,23 @@ keyspace_and_tables parse_keyspace_and_tables(scylla_rest_client& client, const
     return ret;
 }
 
+keyspace_and_tables parse_keyspace_and_tables(scylla_rest_client& client, const bpo::variables_map& vm) {
+    keyspace_and_tables ret;
+
+    ret.keyspace = vm["keyspace"].as<sstring>();
+
+    const auto all_keyspaces = get_keyspaces(client);
+    if (std::ranges::find(all_keyspaces, ret.keyspace) == all_keyspaces.end()) {
+        throw std::invalid_argument(fmt::format("keyspace {} does not exist", ret.keyspace));
+    }
+
+    if (vm.count("table")) {
+        ret.tables = vm["table"].as<std::vector<sstring>>();
+    }
+
+    return ret;
+}
+
 using operation_func = void(*)(scylla_rest_client&, const bpo::variables_map&);
 
 std::map<operation, operation_func> get_operations_with_func();
@@ -189,6 +206,15 @@ void enablegossip_operation(scylla_rest_client& client, const bpo::variables_map
     client.post("/storage_service/gossiping");
 }
 
+void flush_operation(scylla_rest_client& client, const bpo::variables_map& vm) {
+    const auto [keyspace, tables] = parse_keyspace_and_tables(client, vm);
+    std::unordered_map<sstring, sstring> params;
+    if (!tables.empty()) {
+        params["cf"] = fmt::to_string(fmt::join(tables.begin(), tables.end(), ","));
+    }
+    client.post(format("/storage_service/keyspace_flush/{}", keyspace), std::move(params));
+}
+
 void gettraceprobability_operation(scylla_rest_client& client, const bpo::variables_map& vm) {
     auto res = client.get("/storage_service/trace_probability");
     fmt::print(std::cout, "Current trace probability: {}\n", res.GetDouble());
@@ -427,6 +453,24 @@ Fore more information, see: https://opensource.docs.scylladb.com/stable/operatin
             },
             enablegossip_operation
         },
+        {
+            {
+                "flush",
+                "Flush one or more tables",
+R"(
+Specify a keyspace and one or more tables that you want to flush from the
+memtable to on disk SSTables.
+
+Fore more information, see: https://opensource.docs.scylladb.com/stable/operating-scylla/nodetool-commands/flush.html
+)",
+                { },
+                {
+                    typed_option<sstring>("keyspace", "The keyspace to flush", 1),
+                    typed_option<std::vector<sstring>>("table", "The table(s) to flush", -1),
+                }
+            },
+            flush_operation
+        },
         {
             {
                 "gettraceprobability",