Repository: apache/kafka
File: clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerTest.java (Java)

Code before:
            @ClusterConfigProperty(key = "auto.create.topics.enable", value = "false"),
            @ClusterConfigProperty(key = "group.share.partition.max.record.locks", value = "10000"),
            @ClusterConfigProperty(key = "group.share.record.lock.duration.ms", value = "15000"),
            @ClusterConfigProperty(key = "group.share.delivery.count.limit", value = "10"),
            @ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "3"),
            @ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "3"),
            @ClusterConfigProperty(key = "share.coordinator.state.topic.num.partitions", value = "3"),
            @ClusterConfigProperty(key = "share.coordinator.state.topic.replication.factor", value = "3")
        }
    )
    @Timeout(150)
    public void testComplexShareConsumer() throws Exception {
        String topicName = "multipart";
        String groupId = "multipartGrp";
        createTopic(topicName, 3, 3);
        TopicPartition multiTp = new TopicPartition(topicName, 0);

        ScheduledExecutorService service = Executors.newScheduledThreadPool(5);

        ClientState prodState = new ClientState();

        // Produce a fixed number of messages for deterministic testing.
        int targetRecordCount = 2000;
        service.execute(() -> {
            try (Producer<byte[], byte[]> producer = createProducer()) {
                while (prodState.count().get() < targetRecordCount) {
                    ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(multiTp.topic(), multiTp.partition(), null, "key".getBytes(), "value".getBytes());
                    producer.send(record);
                    producer.flush();
                    prodState.count().incrementAndGet();
                }
                prodState.done().set(true);
            }
        });

        // Init a complex share consumer.
        ComplexShareConsumer<byte[], byte[]> complexCons1 = new ComplexShareConsumer<>(
            cluster.bootstrapServers(),
            topicName,
            groupId,
            Map.of(ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG, EXPLICIT)
        );
        alterShareAutoOffsetReset(groupId, "earliest");

        service.schedule(
            complexCons1,
            100L,
            TimeUnit.MILLISECONDS
        );

        // All messages which can be read are read, some would be redelivered (roughly 2 times the records produced).

Review comment (suggestion):
This could easily be a `do { ... } while(prodState.count().incrementAndGet() < targetRecordCount)`. Doesn't seem necessary to call the methods on the `AtomicInteger` twice per loop iteration.

Code after:
            @ClusterConfigProperty(key = "auto.create.topics.enable", value = "false"),
            @ClusterConfigProperty(key = "group.share.partition.max.record.locks", value = "10000"),
            @ClusterConfigProperty(key = "group.share.record.lock.duration.ms", value = "15000"),
            @ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "3"),
            @ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "3"),
            @ClusterConfigProperty(key = "share.coordinator.state.topic.num.partitions", value = "3"),
            @ClusterConfigProperty(key = "share.coordinator.state.topic.replication.factor", value = "3")
        }
    )
    @Timeout(150)
    public void testComplexShareConsumer() throws Exception {
        String topicName = "multipart";
        String groupId = "multipartGrp";
        createTopic(topicName, 3, 3);
        TopicPartition multiTp = new TopicPartition(topicName, 0);

        ScheduledExecutorService service = Executors.newScheduledThreadPool(5);

        ClientState prodState = new ClientState();

        // Produce a fixed number of messages for deterministic testing.
        int targetRecordCount = 2000;
        service.execute(() -> {
            try (Producer<byte[], byte[]> producer = createProducer()) {
                do {
                    ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(multiTp.topic(), multiTp.partition(), null, "key".getBytes(), "value".getBytes());
                    producer.send(record);
                    producer.flush();
                } while (prodState.count().incrementAndGet() < targetRecordCount);
                prodState.done().set(true);
            }
        });

        // Init a complex share consumer.
        ComplexShareConsumer<byte[], byte[]> complexCons1 = new ComplexShareConsumer<>(
            cluster.bootstrapServers(),
            topicName,
            groupId,
            Map.of(ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG, EXPLICIT)
        );
        alterShareAutoOffsetReset(groupId, "earliest");

        service.schedule(
            complexCons1,
            100L,
            TimeUnit.MILLISECONDS
        );

        // All messages which can be read are read, some would be redelivered (roughly 2 times the records produced).
        TestUtils.waitForCondition(complexCons1::isDone, 45_000L, () -> "did not close!");
        int delta = complexCons1.recordsRead() - (int) (prodState.count().get() * 2 * 0.95);    // 2 times with margin of error (5%).