Joe/temp dc fixture demo

src/v/kafka/client/direct_consumer/direct_consumer.cc

@@ -56,6 +56,7 @@ direct_consumer::update_fetchers(topic_partition_map<subscription> removals) {
         co_return;
     }
     auto holder = _gate.hold();
+    auto subscription_lock_holder = co_await _subscriptions_lock.get_units();
     /**
      * Unassign partitions from fetchers that are no longer needed.
      */
@@ -153,7 +154,12 @@ ss::future<> direct_consumer::start() {
 ss::future<> direct_consumer::stop() {
     _cluster->unregister_metadata_cb(_metadata_callback_id);
     _fetched_data_queue->stop();
-    co_await _gate.close();
+
+    // close the gate and break the lock together, no need to drain all
+    // operations
+    auto f = _gate.close();
+    _subscriptions_lock.broken();
+    co_await std::move(f);
 
     co_await ss::parallel_for_each(
       _broker_fetchers, [](auto& pair) { return pair.second->stop(); });

src/v/kafka/client/direct_consumer/direct_consumer.h

@@ -144,6 +144,8 @@ class direct_consumer {
 
     configuration _config;
 
+    // serialize updates to _subscriptions
+    mutex _subscriptions_lock{"direct_consumer::_subscriptions_lock"};
     topic_partition_map<subscription> _subscriptions;
     chunked_hash_map<model::node_id, std::unique_ptr<fetcher>> _broker_fetchers;
     std::unique_ptr<data_queue> _fetched_data_queue;

src/v/kafka/client/direct_consumer/fetcher.cc

@@ -529,6 +529,9 @@ fetcher::process_fetch_response(
                   part_data.high_watermark,
                   find_assignment_epoch(
                     topic_data.topic, part_data.partition_id, epochs));
+                // if (!updated_offset) {
+                //     continue;
+                // }
                 if (updated_offset) {
                     dirty_partitions[topic_data.topic].insert(
                       part_data.partition_id);