Fixes #24627

Motivation

As shared in #24627, ZkSessionExpireTest is flaky. In the cleanup method, the main thread tries to close a PulsarService but then remains in WAITING state, and the test runs into the one hour timeout.

The logs shared in the issue contain a ConcurrentModificationException:

java.util.ConcurrentModificationException
	at java.base/java.util.ArrayList$Itr.checkForComodification(ArrayList.java:1095)
	at java.base/java.util.ArrayList$Itr.next(ArrayList.java:1049)
	at org.apache.pulsar.broker.service.Ipv4Proxy.disconnectFrontChannels(Ipv4Proxy.java:87)
	at org.apache.pulsar.broker.service.ZkSessionExpireTest.testTopicUnloadAfterSessionRebuild(ZkSessionExpireTest.java:149)

The ZkSessionExpireTest calls Ipv4Proxy.disconnectFrontChannels() to inject a ZK session expire error:

metadataZKProxy.rejectAllConnections();
metadataZKProxy.disconnectFrontChannels();

The method disconnectFrontChannels() looks as follows:

for (Channel channel : frontChannels) {
    channel.close();
}

When channel.close() is called, it triggers FrontendHandler.channelInactive(), which removes the channel from the frontChannels list:

@Override
public void channelInactive(ChannelHandlerContext ctx) {
    frontChannels.remove(ctx.channel());
    ...
}

This can trigger the ConcurrentModificationException.

If the ConcurrentModificationException is thrown, the test execution stops. We don't reach the point of the test where the rejectAllConnections is reverted. Therefore, the cleanup method cannot close the PulsarService properly.

I was able to replicate the ConcurrentModificationException by inserting a Thread.sleep() into disconnectFrontChannels():

public void disconnectFrontChannels() throws InterruptedException {
    for (Channel channel : frontChannels) {
        Thread.sleep(5);
        channel.close();
    }
}

Modifications

Prevent ConcurrentModificationException by iterating over a copy of frontChannels.

Verifying this change

• Make sure that the change passes the CI checks.

When adding the Thread.sleep(), the ConcurrentModificationException was always thrown. After changing the for-loop to iterate over a copy of frontChannels, even with the Thread.sleep(), no ConcurrentModificationException occurs.

Does this pull request potentially affect one of the following parts:

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository: pdolif#13