StopProducing after shutdown

Author: Rafał Hibner

cpp/src/arrow/acero/asof_join_node.cc

@@ -1046,8 +1046,10 @@ class AsofJoinNode : public ExecNode {
           if (st.ok()) {
             st = output_->InputFinished(this, batches_produced_);
           }
-          for (const auto& s : state_) {
+          for (size_t i = 0; i < state_.size(); ++i) {
+            const auto& s = state_[i];
             s->ForceShutdown();
+            st &= inputs_[i]->StopProducing();
           }
         }));
   }
@@ -1499,8 +1501,11 @@ class AsofJoinNode : public ExecNode {
     if (st.ok()) {
       st = output_->InputFinished(this, batches_produced_);
     }
-    for (const auto& s : state_) {
-      st &= s->ForceShutdown();
+
+    for (size_t i = 0; i < state_.size(); ++i) {
+      const auto& s = state_[i];
+      s->ForceShutdown();
+      st &= inputs_[i]->StopProducing();
     }
   }
 

cpp/src/arrow/acero/util_test.cc

@@ -299,7 +299,6 @@ TEST(BackpressureConcurrentQueue, BackpressureTest) {
   ASSERT_FALSE(dummy_node.stopped);
   queue.ForceShutdown();
   ASSERT_FALSE(dummy_node.paused);
-  ASSERT_TRUE(dummy_node.stopped);
 }
 
 }  // namespace acero