Re: [PR] KAFKA-20245: DLQ client REJECT ack records. [2/N] [kafka]

["smjn (via GitHub)" <gi...@apache.org>]

smjn commented on PR #21793:
URL: https://github.com/apache/kafka/pull/21793#issuecomment-4222365131

   ### Manual tests
   Broker
   ```
   [2026-04-10 14:24:03,674] WARN Enqueuing share group dlq record parameter: ShareGroupDLQRecordParameter[groupId=gs1, topicIdPartition=zkApK4awQfuB16QwhQKrlg:t1-0, firstOffset=4, lastOffset=4, deliveryCount=Optional[1], cause=Optional[org.apache.kafka.server.share.dlq.ShareGroupDLQ$ShareGroupDLQThrowable: Offset rejected by client.], preserveRecordData=false] (org.apache.kafka.server.share.dlq.NoOpShareGroupDLQManager)
   ```
   
   Client
   ```
   Accept or Reject or Exit (1/2/0)? 1
   Accept or Reject or Exit (1/2/0)? 2
   ```
   
   Interactive client snippet
   ```
   package org.apache.kafka.tools.consumer;
   
   import org.apache.kafka.clients.consumer.AcknowledgeType;
   import org.apache.kafka.clients.consumer.ConsumerRecord;
   import org.apache.kafka.clients.consumer.ConsumerRecords;
   import org.apache.kafka.clients.consumer.KafkaShareConsumer;
   import org.apache.kafka.clients.consumer.ShareConsumer;
   
   import java.time.Duration;
   import java.util.List;
   import java.util.Properties;
   import java.util.Scanner;
   
   public class ShareConsumerDLQ {
   
       private final Scanner scanner = new Scanner(System.in);
   
       public void start(String[] args) {
           Properties props = new Properties();
           props.put("bootstrap.servers", args[0]);
           props.put("share.acknowledgement.mode", args[1]);
           props.put("group.id", args[2]);
           props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
           props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
           try (ShareConsumer<String, String> consumer = new KafkaShareConsumer<>(props)) {
               consumer.subscribe(List.of(args[3]));
               consumer.setAcknowledgementCommitCallback(((offsets, exp) -> {
                   if (exp != null) {
                       System.err.println("error acking records " + exp.getMessage());
                   }
               }));
   
               out:
               while (true) {
                   ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                   for (ConsumerRecord<String, String> record : records) {
                       System.out.print("Accept or Reject or Exit (1/2/0)? ");
                       int choice = scanner.nextInt();
                       if (choice == 0) {
                           break out;
                       }
                       if (choice == 1) {
                           consumer.acknowledge(record, AcknowledgeType.ACCEPT);
                       } else {
                           consumer.acknowledge(record, AcknowledgeType.REJECT);
                       }
                   }
                   consumer.commitSync();
               }
           }
       }
   
       public static void main(String[] args) {
           new ShareConsumerDLQ().start(args);
       }
   }
   
   ```


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org