KAFKA-15852: Move UncleanLeaderElectionTest to server module

[mimaison]

Convert the test from QuorumTestHarness to @ClusterTest.

Since we can have @ParameterizedTest with @ClusterTest I created
Classic and Consumer protocol versions of all the tests that use a
consumer.

[akhileshchg]

Nit: Doesn't need to be public.

Or we can better move it to utils to be reused.

[akhileshchg]

Same as above.

[akhileshchg]

PR looks great. LGTM with minor comments.

[chia7712]

@mimaison thanks for this migration

[chia7712]

Could you merge pollRecordsUntilTrue with pollUntilAtLeastNumRecords? For example:

    private static <K, V> List<ConsumerRecord<K, V>> pollUntilAtLeastNumRecords(Consumer<K, V> consumer, int numRecords) throws Exception {
        final List<ConsumerRecord<K, V>> records = new ArrayList<>();
        waitForCondition(() -> {
            consumer.poll(Duration.ofMillis(100L)).forEach(records::add);
            return records.size() >= numRecords;
        }, DEFAULT_MAX_WAIT_MS, () -> "Consumed " + records.size() + " records before timeout instead of the expected " + numRecords + " records");
        return records;
    }

[mimaison]

The original methods in kafka.utils.TestUtils are called from multiple places hence why I kept them separate there too.

[chia7712]

    public static int awaitLeaderChange(ClusterInstance cluster, TopicPartition tp, Optional<Integer> expectedLeaderOpt, long timeout) throws InterruptedException {
        if (expectedLeaderOpt.isPresent()) {
            LOG.debug("Waiting for leader of {} to change to {}", tp, expectedLeaderOpt.get());
        } else {
            LOG.debug("Waiting for a leader to be elected for {}", tp);
        }

        Supplier<Optional<Integer>> newLeaderExists = () -> cluster.brokers().values().stream()
            .filter(broker -> expectedLeaderOpt.isEmpty() || broker.config().brokerId() == expectedLeaderOpt.get())
            .filter(broker -> broker.replicaManager().onlinePartition(tp).exists(partition -> partition.leaderLogIfLocal().isDefined()))
            .map(broker -> broker.config().brokerId())
            .findFirst();

        waitForCondition(
            () -> newLeaderExists.get().isPresent(),
            timeout,
            "Did not observe leader change for partition " + tp + " after " + timeout + " ms"
        );

        return newLeaderExists.get().get();
    }

[chia7712]

    private void waitForNoLeaderAndIsrHasOldLeaderId(MetadataCache metadataCache, int leaderId) throws InterruptedException {
        waitForCondition(
            () -> metadataCache.getLeaderAndIsr(TOPIC, PARTITION_ID)
                    .filter(leaderAndIsr -> leaderAndIsr.leader() == LeaderConstants.NO_LEADER)
                    .filter(leaderAndIsr -> leaderAndIsr.isr().equals(Set.of(leaderId)))
                    .isPresent(),
                DEFAULT_MAX_WAIT_MS,
                "Timed out waiting for broker metadata cache updates the info for topic partition:" + TOPIC_PARTITION);
    }

[chia7712]

        var configEntries = configs.entrySet().stream()
            .map(e -> new AlterConfigOp(new ConfigEntry(e.getKey(), e.getValue()),
                    AlterConfigOp.OpType.SET))
            .toList();

We could use a single map operation to create the AlterConfigOp instances

[mimaison]

Good idea, I pushed an update. Thanks

[chia7712]

It seems the previous behaviour was to generate a different topic name for each test case

[mimaison]

Each test uses a new cluster, so instead I was thinking of removing the randomness altogether to make things simpler.