diff --git a/cpp/src/arrow/acero/asof_join_node_test.cc b/cpp/src/arrow/acero/asof_join_node_test.cc
index 59a9b4ebba12..bfb08d01a5ee 100644
--- a/cpp/src/arrow/acero/asof_join_node_test.cc
+++ b/cpp/src/arrow/acero/asof_join_node_test.cc
@@ -1373,56 +1373,6 @@ TRACED_TEST(AsofJoinTest, TestUnorderedOnKey, {
       schema({field("time", int64()), field("key", int32()), field("r0_v0", float64())}));
 })
 
-struct BackpressureCounters {
-  std::atomic<int32_t> pause_count = 0;
-  std::atomic<int32_t> resume_count = 0;
-};
-
-struct BackpressureCountingNodeOptions : public ExecNodeOptions {
-  BackpressureCountingNodeOptions(BackpressureCounters* counters) : counters(counters) {}
-
-  BackpressureCounters* counters;
-};
-
-struct BackpressureCountingNode : public MapNode {
-  static constexpr const char* kKindName = "BackpressureCountingNode";
-  static constexpr const char* kFactoryName = "backpressure_count";
-
-  static void Register() {
-    auto exec_reg = default_exec_factory_registry();
-    if (!exec_reg->GetFactory(kFactoryName).ok()) {
-      ASSERT_OK(exec_reg->AddFactory(kFactoryName, BackpressureCountingNode::Make));
-    }
-  }
-
-  BackpressureCountingNode(ExecPlan* plan, std::vector<ExecNode*> inputs,
-                           std::shared_ptr<Schema> output_schema,
-                           const BackpressureCountingNodeOptions& options)
-      : MapNode(plan, inputs, output_schema), counters(options.counters) {}
-
-  static Result<ExecNode*> Make(ExecPlan* plan, std::vector<ExecNode*> inputs,
-                                const ExecNodeOptions& options) {
-    RETURN_NOT_OK(ValidateExecNodeInputs(plan, inputs, 1, kKindName));
-    auto bp_options = static_cast<const BackpressureCountingNodeOptions&>(options);
-    return plan->EmplaceNode<BackpressureCountingNode>(
-        plan, inputs, inputs[0]->output_schema(), bp_options);
-  }
-
-  const char* kind_name() const override { return kKindName; }
-  Result<ExecBatch> ProcessBatch(ExecBatch batch) override { return batch; }
-
-  void PauseProducing(ExecNode* output, int32_t counter) override {
-    ++counters->pause_count;
-    inputs()[0]->PauseProducing(this, counter);
-  }
-  void ResumeProducing(ExecNode* output, int32_t counter) override {
-    ++counters->resume_count;
-    inputs()[0]->ResumeProducing(this, counter);
-  }
-
-  BackpressureCounters* counters;
-};
-
 AsyncGenerator<std::optional<ExecBatch>> GetGen(
     AsyncGenerator<std::optional<ExecBatch>> gen) {
   return gen;
@@ -1453,8 +1403,7 @@ void TestBackpressure(BatchesMaker maker, int batch_size, int num_l_batches,
   ASSERT_OK_AND_ASSIGN(auto r0_batches, make_shift(num_r0_batches, r0_schema, 1));
   ASSERT_OK_AND_ASSIGN(auto r1_batches, make_shift(num_r1_batches, r1_schema, 2));
 
-  BackpressureCountingNode::Register();
-  RegisterTestNodes();  // for GatedNode
+  RegisterTestNodes();  // for GatedNode and BackpressureCountingNode
 
   struct BackpressureSourceConfig {
     std::string name_prefix;
@@ -1499,7 +1448,7 @@ void TestBackpressure(BatchesMaker maker, int batch_size, int num_l_batches,
         std::make_shared<BackpressureCountingNodeOptions>(&bp_counters[i]));
     std::shared_ptr<ExecNodeOptions> options = bp_options.back();
     std::vector<Declaration::Input> bp_in = {src_decls.back()};
-    Declaration bp_decl = {BackpressureCountingNode::kFactoryName, bp_in,
+    Declaration bp_decl = {BackpressureCountingNodeOptions::kName, bp_in,
                            std::move(options)};
     if (config.is_gated) {
       bp_decl = {std::string{GatedNodeOptions::kName}, {bp_decl}, gate_options};
diff --git a/cpp/src/arrow/acero/sorted_merge_node.cc b/cpp/src/arrow/acero/sorted_merge_node.cc
index 37997232cd04..90025da736ab 100644
--- a/cpp/src/arrow/acero/sorted_merge_node.cc
+++ b/cpp/src/arrow/acero/sorted_merge_node.cc
@@ -17,13 +17,16 @@
 
 #include <any>
 #include <atomic>
+#include <memory>
 #include <mutex>
 #include <sstream>
 #include <thread>
 #include <tuple>
 #include <unordered_map>
+#include <utility>
 #include <vector>
 
+#include "arrow/acero/accumulation_queue.h"
 #include "arrow/acero/concurrent_queue_internal.h"
 #include "arrow/acero/exec_plan.h"
 #include "arrow/acero/exec_plan_internal.h"
@@ -34,6 +37,7 @@
 #include "arrow/acero/util.h"
 #include "arrow/array/builder_base.h"
 #include "arrow/result.h"
+#include "arrow/testing/process.h"
 #include "arrow/type_fwd.h"
 #include "arrow/util/logging_internal.h"
 
@@ -99,7 +103,7 @@ class BackpressureController : public BackpressureControl {
 
 /// InputState corresponds to an input. Input record batches are queued up in InputState
 /// until processed and turned into output record batches.
-class InputState {
+class InputState : public util::SerialSequencingQueue::Processor {
  public:
   InputState(size_t index, BackpressureHandler handler,
              const std::shared_ptr<arrow::Schema>& schema, const int time_col_index)
@@ -107,7 +111,9 @@ class InputState {
         queue_(std::move(handler)),
         schema_(schema),
         time_col_index_(time_col_index),
-        time_type_id_(schema_->fields()[time_col_index_]->type()->id()) {}
+        time_type_id_(schema_->fields()[time_col_index_]->type()->id()) {
+    sequencer_ = util::SerialSequencingQueue::Make(this);
+  }
 
   template <typename PtrType>
   static arrow::Result<PtrType> Make(size_t index, arrow::acero::ExecNode* input,
@@ -211,6 +217,15 @@ class InputState {
     return arrow::Status::OK();
   }
 
+  Status InsertBatch(ExecBatch batch) {
+    return sequencer_->InsertBatch(std::move(batch));
+  }
+
+  arrow::Status Process(arrow::ExecBatch batch) final {
+    ARROW_ASSIGN_OR_RAISE(std::shared_ptr<RecordBatch> rb, batch.ToRecordBatch(schema_));
+    return Push(rb);
+  }
+
   const std::shared_ptr<arrow::Schema>& get_schema() const { return schema_; }
 
   void set_total_batches(int n) { total_batches_ = n; }
@@ -219,6 +234,7 @@ class InputState {
   size_t index_;
   // Pending record batches. The latest is the front. Batches cannot be empty.
   BackpressureConcurrentQueue<std::shared_ptr<arrow::RecordBatch>> queue_;
+  std::unique_ptr<util::SerialSequencingQueue> sequencer_;
   // Schema associated with the input
   std::shared_ptr<arrow::Schema> schema_;
   // Total number of batches (only int because InputFinished uses int)
@@ -301,6 +317,9 @@ class SortedMergeNode : public ExecNode {
             "was: ",
             schema->ToString(), " got schema: ", input->output_schema()->ToString());
       }
+      if (input->ordering().is_unordered()) {
+        return Status::Invalid("Input have to be ordered");
+      }
     }
 
     const auto& order_options =
@@ -322,16 +341,16 @@ class SortedMergeNode : public ExecNode {
   arrow::Status Init() override {
     ARROW_CHECK(ordering_.sort_keys().size() == 1) << "Only one sort key supported";
 
+    const auto& sort_key = ordering_.sort_keys()[0];
+    if (sort_key.order != arrow::compute::SortOrder::Ascending) {
+      return Status::NotImplemented("Only ascending sort order is supported");
+    }
+
     auto inputs = this->inputs();
     for (size_t i = 0; i < inputs.size(); i++) {
       ExecNode* input = inputs[i];
       const auto& schema = input->output_schema();
 
-      const auto& sort_key = ordering_.sort_keys()[0];
-      if (sort_key.order != arrow::compute::SortOrder::Ascending) {
-        return Status::NotImplemented("Only ascending sort order is supported");
-      }
-
       const FieldRef& ref = sort_key.target;
       auto match_res = ref.FindOne(*schema);
       if (!match_res.ok()) {
@@ -353,13 +372,12 @@ class SortedMergeNode : public ExecNode {
                               arrow::ExecBatch batch) override {
     ARROW_DCHECK(std_has(inputs_, input));
     const size_t index = std_find(inputs_, input) - inputs_.begin();
-    ARROW_ASSIGN_OR_RAISE(std::shared_ptr<RecordBatch> rb,
-                          batch.ToRecordBatch(output_schema_));
 
-    // Push into the queue. Note that we don't need to lock since
+    // Sequencer menages incoming batches order. then pushes it into
+    // the queue. Note that we don't need to lock since
     // InputState's ConcurrentQueue manages locking
-    input_counter[index] += rb->num_rows();
-    ARROW_RETURN_NOT_OK(state[index]->Push(rb));
+    input_counter[index] += batch.length;
+    ARROW_RETURN_NOT_OK(state[index]->InsertBatch(std::move(batch)));
     PushTask(kNewTask);
     return Status::OK();
   }
@@ -406,6 +424,24 @@ class SortedMergeNode : public ExecNode {
     return Status::OK();
   }
 
+  arrow::Status StopProducing() override {
+#ifdef ARROW_ENABLE_THREADING
+    Future<> to_finish;
+    {
+      std::lock_guard<std::mutex> lg(backpressure_mutex_);
+      if (!backpressure_future_.is_finished()) {
+        to_finish = backpressure_future_;
+        backpressure_future_ = Future<>::MakeFinished();
+      }
+    }
+    if (to_finish.is_valid()) {
+      to_finish.MarkFinished();
+    }
+#endif
+
+    return ExecNode::StopProducing();
+  }
+
   arrow::Status StopProducingImpl() override {
 #ifdef ARROW_ENABLE_THREADING
     process_queue.Clear();
@@ -415,8 +451,39 @@ class SortedMergeNode : public ExecNode {
   }
 
   // handled by the backpressure controller
-  void PauseProducing(arrow::acero::ExecNode* output, int32_t counter) override {}
-  void ResumeProducing(arrow::acero::ExecNode* output, int32_t counter) override {}
+  void PauseProducing(ExecNode* output, int32_t counter) override {
+#ifdef ARROW_ENABLE_THREADING
+    std::lock_guard<std::mutex> lg(backpressure_mutex_);
+    if (counter <= last_backpressure_counter_) {
+      return;
+    }
+    last_backpressure_counter_ = counter;
+    if (!backpressure_future_.is_finished()) {
+      // Could happen if we get something like Pause(1) Pause(3) Resume(2)
+      return;
+    }
+    backpressure_future_ = Future<>::Make();
+#endif
+  }
+
+  void ResumeProducing(ExecNode* output, int32_t counter) override {
+#ifdef ARROW_ENABLE_THREADING
+    Future<> to_finish;
+    {
+      std::lock_guard<std::mutex> lg(backpressure_mutex_);
+      if (counter <= last_backpressure_counter_) {
+        return;
+      }
+      last_backpressure_counter_ = counter;
+      if (backpressure_future_.is_finished()) {
+        return;
+      }
+      to_finish = backpressure_future_;
+      backpressure_future_ = Future<>::MakeFinished();
+    }
+    to_finish.MarkFinished();
+#endif
+  }
 
  protected:
   std::string ToStringExtra(int indent) const override {
@@ -587,6 +654,13 @@ class SortedMergeNode : public ExecNode {
 #ifdef ARROW_ENABLE_THREADING
   void EmitBatches() {
     while (true) {
+      Future<> to_wait;
+      {
+        std::lock_guard<std::mutex> lg(backpressure_mutex_);
+        to_wait = backpressure_future_;
+      }
+      to_wait.Wait();
+
       // Implementation note: If the queue is empty, we will block here
       if (process_queue.WaitAndPop() == kPoisonPill) {
         EndFromProcessThread();
@@ -613,7 +687,7 @@ class SortedMergeNode : public ExecNode {
   std::atomic<bool> cleanup_started{false};
 
   // Backpressure counter common to all input states
-  std::atomic<int32_t> backpressure_counter;
+  std::atomic<int32_t> backpressure_counter{0};
 
   std::atomic<int32_t> batches_produced{0};
 
@@ -623,6 +697,10 @@ class SortedMergeNode : public ExecNode {
   // Once StartProducing is called, we initialize this thread to poll the
   // input states and emit batches
   std::thread process_thread;
+
+  std::mutex backpressure_mutex_;
+  std::atomic<int32_t> last_backpressure_counter_{0};
+  Future<> backpressure_future_ = Future<>::MakeFinished();
 #endif
   arrow::Future<> process_task;
 
@@ -642,4 +720,4 @@ void RegisterSortedMergeNode(ExecFactoryRegistry* registry) {
 }
 }  // namespace internal
 
-}  // namespace arrow::acero
+}  // namespace arrow::acero
\ No newline at end of file
diff --git a/cpp/src/arrow/acero/sorted_merge_node_test.cc b/cpp/src/arrow/acero/sorted_merge_node_test.cc
index 82b630420c4a..a1a9db9fe1c3 100644
--- a/cpp/src/arrow/acero/sorted_merge_node_test.cc
+++ b/cpp/src/arrow/acero/sorted_merge_node_test.cc
@@ -23,12 +23,17 @@
 #include "arrow/acero/test_nodes.h"
 #include "arrow/array/builder_base.h"
 #include "arrow/array/concatenate.h"
+#include "arrow/compute/cast.h"
 #include "arrow/compute/ordering.h"
+#include "arrow/compute/row/row_encoder_internal.h"
+#include "arrow/compute/test_util_internal.h"
+#include "arrow/io/util_internal.h"
 #include "arrow/result.h"
 #include "arrow/scalar.h"
 #include "arrow/table.h"
 #include "arrow/testing/generator.h"
 #include "arrow/testing/gtest_util.h"
+#include "arrow/testing/matchers.h"
 #include "arrow/type.h"
 #include "arrow/type_fwd.h"
 
@@ -65,8 +70,9 @@ TEST(SortedMergeNode, Basic) {
   auto ops = OrderByNodeOptions(compute::Ordering({compute::SortKey("timestamp")}));
 
   Declaration sorted_merge{"sorted_merge", src_decls, ops};
-  // We can't use threads for sorted merging since it relies on
-  // ascending deterministic order of timestamps
+  // Now We can use threads for sorted merging since now it uses sequencer to assure
+  // deterministic order of timestamps, but for the sake of not modifying this test we
+  // won't use threads
   ASSERT_OK_AND_ASSIGN(auto output,
                        DeclarationToTable(sorted_merge, /*use_threads=*/false));
   ASSERT_EQ(output->num_rows(), 18);
@@ -83,4 +89,193 @@ TEST(SortedMergeNode, Basic) {
   AssertArraysEqual(*expected_ts, *output_ts);
 }
 
-}  // namespace arrow::acero
+TEST(SortedMergeNode, BasicThreaded) {
+  auto table1 = TestTable(
+      /*start=*/0,
+      /*step=*/2,
+      /*rows_per_batch=*/2,
+      /*num_batches=*/3);
+  auto table2 = TestTable(
+      /*start=*/1,
+      /*step=*/2,
+      /*rows_per_batch=*/3,
+      /*num_batches=*/2);
+  auto table3 = TestTable(
+      /*start=*/3,
+      /*step=*/3,
+      /*rows_per_batch=*/6,
+      /*num_batches=*/1);
+  static constexpr random::SeedType kTestSeed = 42;
+  static constexpr int kMaxJitterMod = 2;
+  RegisterTestNodes();
+  std::vector<Declaration::Input> src_decls;
+  src_decls.emplace_back(
+      Declaration("jitter", {Declaration("table_source", TableSourceNodeOptions(table1))},
+                  JitterNodeOptions(kTestSeed, kMaxJitterMod)));
+  src_decls.emplace_back(
+      Declaration("jitter", {Declaration("table_source", TableSourceNodeOptions(table2))},
+                  JitterNodeOptions(kTestSeed, kMaxJitterMod)));
+  src_decls.emplace_back(
+      Declaration("jitter", {Declaration("table_source", TableSourceNodeOptions(table3))},
+                  JitterNodeOptions(kTestSeed, kMaxJitterMod)));
+
+  auto ops = OrderByNodeOptions(compute::Ordering({compute::SortKey("timestamp")}));
+  Declaration sorted_merge{"sorted_merge", src_decls, ops};
+  // Now We can use threads for sorted merging since it uses sequencer to assure
+  // deterministic order of timestamps.
+  // To simulate wrong order usually caused by enabling threads we use jitter node which
+  // would previously break the output without using sequencer.
+  ASSERT_OK_AND_ASSIGN(auto output,
+                       DeclarationToTable(sorted_merge, /*use_threads=*/true));
+  ASSERT_EQ(output->num_rows(), 18);
+
+  ASSERT_OK_AND_ASSIGN(auto expected_ts_builder,
+                       MakeBuilder(int32(), default_memory_pool()));
+  for (auto i : {0, 1, 2, 3, 3, 4, 5, 6, 6, 7, 8, 9, 9, 10, 11, 12, 15, 18}) {
+    ASSERT_OK(expected_ts_builder->AppendScalar(*MakeScalar(i)));
+  }
+  ASSERT_OK_AND_ASSIGN(auto expected_ts, expected_ts_builder->Finish());
+  auto output_col = output->column(0);
+  ASSERT_OK_AND_ASSIGN(auto output_ts, Concatenate(output_col->chunks()));
+  AssertArraysEqual(*expected_ts, *output_ts);
+}
+
+TEST(SortedMergeNode, PauseProducingSortedMerge) {
+#ifndef ARROW_ENABLE_THREADING
+  GTEST_SKIP() << "Test requires threading support";
+#endif
+
+  RegisterTestNodes();
+
+  int batch_size = 1;
+  auto make_shift = [batch_size](int num_batches, const std::shared_ptr<Schema>& schema,
+                                 int shift) {
+    return MakeIntegerBatches(
+        {[](int row) -> int64_t { return row; },
+         [num_batches](int row) -> int64_t { return row / num_batches; },
+         [shift](int row) -> int64_t { return row * 10 + shift; }},
+        schema, num_batches, batch_size);
+  };
+  auto l_schema =
+      schema({field("time", int64()), field("key", int64()), field("l_value", int64())});
+  auto r_schema =
+      schema({field("time", int64()), field("key", int64()), field("l_value", int64())});
+
+  auto output_schema =
+      schema({field("time", int64()), field("key", int64()), field("l_value", int64()),
+              field("key", int64()), field("l_value", int64())});
+
+  ASSERT_OK_AND_ASSIGN(auto out_batch,
+                       MakeIntegerBatches({[](int row) -> int64_t { return row; },
+                                           [](int row) -> int64_t { return row; },
+                                           [](int row) -> int64_t { return row / 20; },
+                                           [](int row) -> int64_t { return row / 20; },
+                                           [](int row) -> int64_t { return row * 10; }},
+                                          output_schema, 20, batch_size))
+
+  ASSERT_OK_AND_ASSIGN(auto l_batches, make_shift(50, l_schema, 2));
+  ASSERT_OK_AND_ASSIGN(auto r0_batches, make_shift(50, r_schema, 1));
+  std::optional<ExecBatch> out = out_batch.batches[0];
+
+  constexpr uint32_t thresholdOfBackpressureSmn = 8;
+
+  EXPECT_OK_AND_ASSIGN(std::shared_ptr<ExecPlan> plan, ExecPlan::Make());
+  PushGenerator<std::optional<ExecBatch>> batch_producer_left;
+  PushGenerator<std::optional<ExecBatch>> batch_producer_right;
+
+  auto ordering = compute::Ordering({compute::SortKey("time")});
+
+  AsyncGenerator<std::optional<ExecBatch>> sink_gen;
+  BackpressureMonitor* backpressure_monitor;
+  BackpressureOptions backpressure_options(1, 2);
+
+  Declaration left{"source", SourceNodeOptions(l_schema, batch_producer_left, ordering)};
+  Declaration right{"source",
+                    SourceNodeOptions(r_schema, batch_producer_right, ordering)};
+
+  auto opt = OrderByNodeOptions(ordering);
+
+  BackpressureCounters bp_countersl, bp_countersr;
+
+  Declaration left_count{"backpressure_count",
+                         {std::move(left)},
+                         BackpressureCountingNodeOptions(&bp_countersl)};
+
+  Declaration right_count{"backpressure_count",
+                          {std::move(right)},
+                          BackpressureCountingNodeOptions(&bp_countersr)};
+
+  Declaration asof_join{
+      "sorted_merge", {std::move(left_count), std::move(right_count)}, std::move(opt)};
+
+  ARROW_EXPECT_OK(
+      acero::Declaration::Sequence(
+          {
+              std::move(asof_join),
+              {"sink", SinkNodeOptions{&sink_gen, /*schema=*/nullptr,
+                                       backpressure_options, &backpressure_monitor}},
+          })
+          .AddToPlan(plan.get()));
+
+  ASSERT_TRUE(backpressure_monitor);
+  plan->StartProducing();
+  auto fut = plan->finished();
+
+  EXPECT_FALSE(backpressure_monitor->is_paused());
+
+  // Should be able to push kPauseIfAbove batches without triggering back pressure
+  int64_t l_cnt = 0;
+  int64_t r_cnt = 0;
+
+  EXPECT_FALSE(bp_countersl.is_paused());
+  EXPECT_FALSE(bp_countersr.is_paused());
+  EXPECT_FALSE(backpressure_monitor->is_paused());
+  batch_producer_left.producer().Push(l_batches.batches[l_cnt++]);
+  batch_producer_right.producer().Push(r0_batches.batches[r_cnt++]);
+  // this should trigger pause on sink
+  BusyWait(3.0, [&]() { return backpressure_monitor->is_paused(); });
+  arrow::io::internal::GetIOThreadPool()->WaitForIdle();
+  arrow::internal::GetCpuThreadPool()->WaitForIdle();
+
+  // Fill up the inputs of the asof join node
+  for (uint32_t i = 0; i < thresholdOfBackpressureSmn; i++) {
+    SleepABit();
+    // EXPECT_FALSE(bp_countersl.is_paused());
+    // EXPECT_FALSE(bp_countersr.is_paused());
+    EXPECT_TRUE(backpressure_monitor->is_paused());
+    batch_producer_left.producer().Push(l_batches.batches[l_cnt++]);
+    batch_producer_right.producer().Push(r0_batches.batches[r_cnt++]);
+  }
+  SleepABit();
+  BusyWait(3.0, [&]() { return bp_countersl.is_paused(); });
+  BusyWait(3.0, [&]() { return bp_countersr.is_paused(); });
+  arrow::io::internal::GetIOThreadPool()->WaitForIdle();
+  arrow::internal::GetCpuThreadPool()->WaitForIdle();
+  SleepABit();
+  // Verify pause propagates
+  EXPECT_TRUE(bp_countersl.is_paused());
+  EXPECT_TRUE(bp_countersr.is_paused());
+
+  std::optional<ExecBatch> opt_batch;
+
+  do {
+    ASSERT_FINISHES_OK_AND_ASSIGN(opt_batch, sink_gen());
+    ASSERT_TRUE(opt_batch);
+    l_cnt -= opt_batch->length;
+  } while (l_cnt > 0);
+
+  BusyWait(3.0, [&]() { return !bp_countersl.is_paused(); });
+  BusyWait(3.0, [&]() { return !bp_countersr.is_paused(); });
+  arrow::io::internal::GetIOThreadPool()->WaitForIdle();
+  arrow::internal::GetCpuThreadPool()->WaitForIdle();
+  EXPECT_FALSE(bp_countersl.is_paused());
+  EXPECT_FALSE(bp_countersr.is_paused());
+  EXPECT_FALSE(backpressure_monitor->is_paused());
+
+  batch_producer_left.producer().Push(IterationEnd<std::optional<ExecBatch>>());
+  batch_producer_right.producer().Push(IterationEnd<std::optional<ExecBatch>>());
+
+  ASSERT_THAT(fut, Finishes(Ok()));
+}
+
+}  // namespace arrow::acero
\ No newline at end of file
diff --git a/cpp/src/arrow/acero/test_nodes.cc b/cpp/src/arrow/acero/test_nodes.cc
index d95c22ca4510..52bb77a96f2b 100644
--- a/cpp/src/arrow/acero/test_nodes.cc
+++ b/cpp/src/arrow/acero/test_nodes.cc
@@ -16,18 +16,18 @@
 // under the License.
 
 #include "arrow/acero/test_nodes.h"
-
 #include <deque>
 #include <iostream>
 #include <mutex>
 #include <string>
 #include <vector>
-
 #include "arrow/acero/exec_plan.h"
+#include "arrow/acero/map_node.h"
 #include "arrow/acero/query_context.h"
 #include "arrow/acero/util.h"
 #include "arrow/api.h"
 #include "arrow/io/interfaces.h"
+#include "arrow/io/util_internal.h"
 #include "arrow/testing/random.h"
 #include "arrow/util/checked_cast.h"
 #include "arrow/util/iterator.h"
@@ -356,6 +356,56 @@ struct GatedNode : public ExecNode, public TracedNode {
 
 }  // namespace
 
+namespace {
+
+struct BackpressureCountingNode : public MapNode {
+  static constexpr const char* kKindName = "BackpressureCountingNode";
+
+  BackpressureCountingNode(ExecPlan* plan, std::vector<ExecNode*> inputs,
+                           std::shared_ptr<Schema> output_schema,
+                           const BackpressureCountingNodeOptions& options)
+      : MapNode(plan, inputs, output_schema), counters(options.counters) {}
+
+  static Result<ExecNode*> Make(ExecPlan* plan, std::vector<ExecNode*> inputs,
+                                const ExecNodeOptions& options) {
+    RETURN_NOT_OK(ValidateExecNodeInputs(plan, inputs, 1, kKindName));
+    auto bp_options = static_cast<const BackpressureCountingNodeOptions&>(options);
+    return plan->EmplaceNode<BackpressureCountingNode>(
+        plan, inputs, inputs[0]->output_schema(), bp_options);
+  }
+
+  const char* kind_name() const override { return kKindName; }
+  Result<ExecBatch> ProcessBatch(ExecBatch batch) override { return batch; }
+
+  void PauseProducing(ExecNode* output, int32_t counter) override {
+    std::lock_guard<std::mutex> lg(mutex_);
+    if (counter > backpressure_counter_) {
+      backpressure_counter_ = counter;
+      ++counters->pause_count;
+      // if (counters->paused)
+      counters->paused = true;
+    }
+    inputs()[0]->PauseProducing(this, counter);
+  }
+
+  void ResumeProducing(ExecNode* output, int32_t counter) override {
+    std::lock_guard<std::mutex> lg(mutex_);
+    if (counter > backpressure_counter_) {
+      backpressure_counter_ = counter;
+      ++counters->resume_count;
+      // if (!counters->paused)
+      counters->paused = false;
+    }
+    inputs()[0]->ResumeProducing(this, counter);
+  }
+
+  BackpressureCounters* counters;
+  std::mutex mutex_;
+  std::atomic<int32_t> backpressure_counter_{0};
+};
+
+}  // namespace
+
 void RegisterTestNodes() {
   static std::once_flag registered;
   std::call_once(registered, [] {
@@ -364,6 +414,8 @@ void RegisterTestNodes() {
         registry->AddFactory(std::string(JitterNodeOptions::kName), JitterNode::Make));
     DCHECK_OK(
         registry->AddFactory(std::string(GatedNodeOptions::kName), GatedNode::Make));
+    DCHECK_OK(registry->AddFactory(std::string(BackpressureCountingNodeOptions::kName),
+                                   BackpressureCountingNode::Make));
   });
 }
 
diff --git a/cpp/src/arrow/acero/test_nodes.h b/cpp/src/arrow/acero/test_nodes.h
index 7e31aa31b34d..634f6df86c84 100644
--- a/cpp/src/arrow/acero/test_nodes.h
+++ b/cpp/src/arrow/acero/test_nodes.h
@@ -53,6 +53,20 @@ struct JitterNodeOptions : public ExecNodeOptions {
   static constexpr std::string_view kName = "jitter";
 };
 
+struct BackpressureCounters {
+  std::atomic<int32_t> pause_count = 0;
+  std::atomic<int32_t> resume_count = 0;
+  std::atomic<bool> paused = false;
+
+  bool is_paused() { return paused; }
+};
+
+struct BackpressureCountingNodeOptions : public ExecNodeOptions {
+  BackpressureCountingNodeOptions(BackpressureCounters* counters) : counters(counters) {}
+  BackpressureCounters* counters;
+  static constexpr const char* kName = "backpressure_count";
+};
+
 class GateImpl;
 
 class Gate {