[jira] [Created] (KAFKA-16194) KafkaConsumer.groupMetadata() should be correct when first records are returned

[David Jacot (Jira)]

David Jacot created KAFKA-16194:
-----------------------------------

             Summary: KafkaConsumer.groupMetadata() should be correct when 
first records are returned
                 Key: KAFKA-16194
                 URL: https://issues.apache.org/jira/browse/KAFKA-16194
             Project: Kafka
          Issue Type: Sub-task
            Reporter: David Jacot


The following code returns records before the group metadata is updated. This 
fails the first transactions ever run by the Producer/Consumer.

 
{code:java}
Producer<String, String> txnProducer = new KafkaProducer<>(txnProducerProps);
Consumer<String, String> consumer = new KafkaConsumer<>(consumerProps);

txnProducer.initTransactions();
System.out.println("Init transactions called");

try {
    txnProducer.beginTransaction();
    System.out.println("Begin transactions called");

    consumer.subscribe(Collections.singletonList("input"));
    System.out.println("Consumer subscribed to topic -> KIP848-topic-2 ");

    ConsumerRecords<String, String> records = 
consumer.poll(Duration.ofSeconds(10));
    System.out.println("Returned " + records.count() + " records.");

    // Process and send txn messages.
    for (ConsumerRecord<String, String> processedRecord : records) {
        txnProducer.send(new ProducerRecord<>("output", processedRecord.key(), 
"Processed: " + processedRecord.value()));
    }

    ConsumerGroupMetadata groupMetadata = consumer.groupMetadata();
    System.out.println("Group metadata inside test" + groupMetadata);

    Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>();
    for (ConsumerRecord<String, String> record : records) {
        offsetsToCommit.put(new TopicPartition(record.topic(), 
record.partition()),
            new OffsetAndMetadata(record.offset() + 1));
    }
    System.out.println("Offsets to commit" + offsetsToCommit);
    // Send offsets to transaction with ConsumerGroupMetadata.
    txnProducer.sendOffsetsToTransaction(offsetsToCommit, groupMetadata);
    System.out.println("Send offsets to transaction done");

    // Commit the transaction.
    txnProducer.commitTransaction();
    System.out.println("Commit transaction done");
} catch (ProducerFencedException | OutOfOrderSequenceException | 
AuthorizationException e) {
    e.printStackTrace();
    txnProducer.close();
} catch (KafkaException e) {
    e.printStackTrace();
    txnProducer.abortTransaction();
} finally {
    txnProducer.close();
    consumer.close();
} {code}
The issue seems to be that while it waits in `poll`, the event to update the 
group metadata is not processed.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)