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

Code before:
            cluster.shutdownBroker(curGroupCoordNodeId.get(0));
            // Wait for it to be completely shutdown
            broker.awaitShutdown();
            // Wait for the leaders of share coordinator, group coordinator and topic partition to be elected, if needed, on a different broker.
            TestUtils.waitForCondition(() -> {
                List<Integer> newShareCoordNodeId = topicPartitionLeader(adminClient, Topic.SHARE_GROUP_STATE_TOPIC_NAME, shareGroupStateTp);
                List<Integer> newGroupCoordNodeId = topicPartitionLeader(adminClient, Topic.GROUP_METADATA_TOPIC_NAME, consumerOffsetsTp);
                List<Integer> newTopicPartitionLeader = topicPartitionLeader(adminClient, tp.topic(), tp.partition());

                return newShareCoordNodeId.size() == 1 && !Objects.equals(newShareCoordNodeId.get(0), curGroupCoordNodeId.get(0)) &&
                    newGroupCoordNodeId.size() == 1 && !Objects.equals(newGroupCoordNodeId.get(0), curGroupCoordNodeId.get(0)) &&
                    newTopicPartitionLeader.size() == 1 && !Objects.equals(newTopicPartitionLeader.get(0), curGroupCoordNodeId.get(0));
            }, DEFAULT_MAX_WAIT_MS, DEFAULT_POLL_INTERVAL_MS, () -> "Failed to elect new leaders after broker shutdown");
            // After group coordinator shutdown, check that lag is still 1
            verifySharePartitionLag(adminClient, groupId, tp, 1L);
        } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException(e);
        }
    }

    @ClusterTest(
        brokers = 3,
        serverProperties = {
            @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")
        }
    )
    public void testSharePartitionLagOnShareCoordinatorMovement() {
        String groupId = "group1";
        alterShareAutoOffsetReset(groupId, "earliest");
        try (Producer<byte[], byte[]> producer = createProducer();
             ShareConsumer<byte[], byte[]> shareConsumer = createShareConsumer(groupId);
             Admin adminClient = createAdminClient()) {
            String topicName = "testTopicWithReplicas";
            // Create a topic with replication factor 3
            Uuid tpId = createTopic(topicName, 1, 3);
            TopicPartition tp = new TopicPartition(topicName, 0);
            ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(tp.topic(), tp.partition(), null, "key".getBytes(), "Message".getBytes());
            // Produce first record and consume it
            producer.send(record);
            producer.flush();
            shareConsumer.subscribe(List.of(tp.topic()));
            // Polling share consumer to make sure the share partition is created and the record is consumed.
            waitedPoll(shareConsumer, 2500L, 1);
            // Acknowledge and commit the consumed record to update the share partition state.
            shareConsumer.commitSync();
            // After the acknowledgement is successful, the share partition lag should be 0 because the only produced record has been consumed.
            verifySharePartitionLag(adminClient, groupId, tp, 0L);
            // Producing another record to the share partition.

Review comment (suggestion):
You do need replication factor 3, but you only need one partition for each of these topics. Then you'd not have to fiddle around trying to calculate the partitions used for the internal topics.

Code after:
            producer.send(record);
            producer.flush();
            shareConsumer.subscribe(List.of(tp.topic()));
            // Polling share consumer to make sure the share partition is created and the record is consumed.
            waitedPoll(shareConsumer, 2500L, 1);
            // Acknowledge and commit the consumed record to update the share partition state.
            shareConsumer.commitSync();
            // After the acknowledgement is successful, the share partition lag should be 0 because the only produced record has been consumed.
            verifySharePartitionLag(adminClient, groupId, tp, 0L);
            // Producing another record to the share partition.
            producer.send(record);
            producer.flush();
            // Since the new record has not been consumed yet, the share partition lag should be 1.
            verifySharePartitionLag(adminClient, groupId, tp, 1L);
            List<Integer> curGroupCoordNodeId;
            // Find the broker which is the group coordinator for the share group.
            curGroupCoordNodeId = topicPartitionLeader(adminClient, Topic.GROUP_METADATA_TOPIC_NAME, 0);
            assertEquals(1, curGroupCoordNodeId.size());
            // Shut down the coordinator broker
            KafkaBroker broker = cluster.brokers().get(curGroupCoordNodeId.get(0));
            cluster.shutdownBroker(curGroupCoordNodeId.get(0));
            // Wait for it to be completely shutdown
            broker.awaitShutdown();
            // Wait for the leaders of share coordinator, group coordinator and topic partition to be elected, if needed, on a different broker.
            TestUtils.waitForCondition(() -> {
                List<Integer> newShareCoordNodeId = topicPartitionLeader(adminClient, Topic.SHARE_GROUP_STATE_TOPIC_NAME, 0);
                List<Integer> newGroupCoordNodeId = topicPartitionLeader(adminClient, Topic.GROUP_METADATA_TOPIC_NAME, 0);
                List<Integer> newTopicPartitionLeader = topicPartitionLeader(adminClient, tp.topic(), tp.partition());

                return newShareCoordNodeId.size() == 1 && !Objects.equals(newShareCoordNodeId.get(0), curGroupCoordNodeId.get(0)) &&
                    newGroupCoordNodeId.size() == 1 && !Objects.equals(newGroupCoordNodeId.get(0), curGroupCoordNodeId.get(0)) &&
                    newTopicPartitionLeader.size() == 1 && !Objects.equals(newTopicPartitionLeader.get(0), curGroupCoordNodeId.get(0));
            }, DEFAULT_MAX_WAIT_MS, DEFAULT_POLL_INTERVAL_MS, () -> "Failed to elect new leaders after broker shutdown");
            // After group coordinator shutdown, check that lag is still 1
            verifySharePartitionLag(adminClient, groupId, tp, 1L);
        } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException(e);
        }
    }

    @ClusterTest(
        brokers = 3,
        serverProperties = {
            @ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
            @ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "3"),
            @ClusterConfigProperty(key = "share.coordinator.state.topic.num.partitions", value = "1"),
            @ClusterConfigProperty(key = "share.coordinator.state.topic.replication.factor", value = "3")
        }
    )
    public void testSharePartitionLagOnShareCoordinatorMovement() {
        String groupId = "group1";