Consume sink batches and clear input

Author: mroz45

cpp/src/arrow/acero/asof_join_node_test.cc

@@ -1683,8 +1683,12 @@ TEST(AsofJoinTest, PauseProducingAsofJoinSource) {
   ASSERT_FINISHES_OK(sink_gen());
   BusyWait(10, [&] { return !backpressure_monitor->is_paused(); });
   ASSERT_FALSE(backpressure_monitor->is_paused());
-
+  
   // Cleanup
+  for (size_t i = 0; i < cnt - 3; i++) {
+    ASSERT_FINISHES_OK(sink_gen());
+  }
+
   batch_producer_left.producer().Push(IterationEnd<std::optional<ExecBatch>>());
   batch_producer_right.producer().Push(IterationEnd<std::optional<ExecBatch>>());