Add explicit mode support for ShareConsumer with alternating commit strategy

Author: Ankith-Confluent

tests/soak/run.sh

@@ -23,12 +23,17 @@ if [[ "${SHARE:-}" == "true" ]]; then
     share_flag="--share"
 fi
 
+explicit_flag=""
+if [[ "${EXPLICIT:-}" == "true" ]]; then
+    explicit_flag="--explicit"
+fi
+
 echo "Starting soak client using topic $topic. Logging to $logfile."
 set +x
 while [ "$run" = true ]; do
     # Ignore SIGINT in children (inherited)
     trap "" SIGINT
-    time opentelemetry-instrument $testdir/soakclient.py -i $TESTID -t $topic -r 80 -f $1 $share_flag |& tee /dev/tty | bzip2 > $logfile &
+    time opentelemetry-instrument $testdir/soakclient.py -i $TESTID -t $topic -r 80 -f $1 $share_flag $explicit_flag |& tee /dev/tty | bzip2 > $logfile &
     PID=$!
     terminate_last() {
         # List children of $PID only

tests/soak/soakclient.py

@@ -368,9 +368,17 @@ def share_run(self):
 
         # Track highest offset seen per partition for duplicate/gap detection.
         # With implicit ack and a single share consumer, offsets should
-        # progress sequentially per partition.
+        # progress sequentially per partition. Skipped in Explict mode.
         hwmarks = defaultdict(int)
 
+        # In explicit mode, commit every N messages and alternate between
+        # commit_async and commit_sync.
+        COMMIT_EVERY_MSGS = 1000
+        msgs_since_commit = 0
+        commit_use_sync = False
+
+        self.logger.info("share: running in mode={}".format(self.share_mode))
+
         next_status = time.time() + self.disprate
 
         while self.run:
@@ -461,6 +469,46 @@ def share_run(self):
 
                 hwmarks[hwkey] = msg.offset()
 
+                # Explicit mode: ack each message with ACCEPT and flush acks
+                # every N messages via a standalone ShareAcknowledge RPC,
+                # alternating commit_async and commit_sync.
+                if self.share_mode == 'explicit':
+                    try:
+                        self.share_consumer.acknowledge(msg)
+                        msgs_since_commit += 1
+                    except KafkaException as ex:
+                        self.logger.error(
+                            "share: acknowledge failed: {}".format(ex))
+                        self.share_err_cnt += 1
+                        self.incr_counter("consumer.error", 1)
+
+                    if msgs_since_commit >= COMMIT_EVERY_MSGS:
+                        try:
+                            if commit_use_sync:
+                                result = self.share_consumer.commit_sync(
+                                    timeout=10.0)
+                                partition_errs = sum(
+                                    1 for err in result.values()
+                                    if err is not None
+                                )
+                                if partition_errs > 0:
+                                    self.logger.warning(
+                                        "share: commit_sync had {} partition "
+                                        "error(s)".format(partition_errs)
+                                    )
+                                    self.share_err_cnt += 1
+                                    self.incr_counter("consumer.error", 1)
+                            else:
+                                self.share_consumer.commit_async()
+                        except KafkaException as ex:
+                            self.logger.error(
+                                "share: commit_{} exception: {}".format(
+                                    "sync" if commit_use_sync else "async", ex))
+                            self.share_err_cnt += 1
+                            self.incr_counter("consumer.error", 1)
+                        commit_use_sync = not commit_use_sync
+                        msgs_since_commit = 0
+
         self.share_consumer.close()
         self.share_status()
 
@@ -541,14 +589,16 @@ def create_topic(self, topic, conf):
             else:
                 raise
 
-    def __init__(self, testid, topic, rate, conf, enable_share=False):
+    def __init__(self, testid, topic, rate, conf, enable_share=False,
+                 share_mode='implicit'):
         """SoakClient constructor. conf is the client configuration"""
         self.topic = topic
         self.rate = rate
         self.disprate = int(rate * 10)
         self.run = True
         self.stats_cnt = {'producer': 0, 'consumer': 0}
         self.start_time = time.time()
+        self.share_mode = share_mode
 
         # OTEL instruments
         self.counters = {}
@@ -637,6 +687,12 @@ def filter_config(conf, filter_out, strip_prefix):
             sconf.pop('stats_cb', None)
             sconf['client.id'] = self.testid
 
+            # In explicit mode, switch the consumer's ack policy. Default is
+            # implicit (next poll auto-acks); without this flip, calls to
+            # acknowledge() return _STATE because the message is already acked.
+            if self.share_mode == 'explicit':
+                sconf['share.acknowledgement.mode'] = 'explicit'
+
             # Always set a share-specific group.id.
             sconf['group.id'] = 'soakclient-share-{}-{}-{}'.format(
                 self.hostname, version(), sys.version.split(' ')[0]
@@ -771,9 +827,18 @@ def get_rusage(self):
         '--share', dest='share', action='store_true', default=False,
         help='Enable share consumer thread'
     )
+    parser.add_argument(
+        '--explicit', dest='explicit', action='store_true', default=False,
+        help='Share consumer: per-msg ACCEPT + alternating commit_async/sync (requires --share)'
+    )
 
     args = parser.parse_args()
 
+    share_mode = 'explicit' if args.explicit else 'implicit'
+
+    if share_mode == 'explicit' and not args.share:
+        parser.error('--explicit requires --share')
+
     conf = dict()
     if args.conffile is not None:
         # Parse client configuration file.
@@ -801,7 +866,8 @@ def get_rusage(self):
     conf['enable.partition.eof'] = False
 
     # Create SoakClient
-    soak = SoakClient(args.testid, args.topic, args.rate, conf, enable_share=args.share)
+    soak = SoakClient(args.testid, args.topic, args.rate, conf,
+                      enable_share=args.share, share_mode=share_mode)
 
     # Get initial resource usage
     soak.get_rusage()