This test failed two times today with a listener being executed twice:
```
rg.elasticsearch.snapshots.ConcurrentSnapshotsIT > testMasterFailoverOnFinalizationLoop FAILED
    com.carrotsearch.randomizedtesting.UncaughtExceptionError: Captured an uncaught exception in thread: Thread[id=7911, name=Thread-51, state=RUNNABLE, group=TGRP-ConcurrentSnapshotsIT]
        at __randomizedtesting.SeedInfo.seed([54D1EB587A8E7FD5:55807F457CEDE6BA]:0)

        Caused by:
        java.lang.AssertionError: org.elasticsearch.ElasticsearchException: org.elasticsearch.action.ActionListenerImplementations$RunBeforeActionListener/ChannelActionListener{TaskTransportChannel{task=73}{TcpTransportChannel{req=33}{cluster:admin/snapshot/delete}{Netty4TcpChannel{localAddress=/127.0.0.1:20923, remoteAddress=/127.0.0.1:45894, profile=default}}}}/org.elasticsearch.action.support.master.TransportMasterNodeAction$$Lambda/0x00007f15e4953ce0@f49da3c
            at __randomizedtesting.SeedInfo.seed([54D1EB587A8E7FD5]:0)
            at org.elasticsearch.action.ActionListener$4.assertFirstRun(ActionListener.java:324)
            at org.elasticsearch.action.ActionListener$4.onFailure(ActionListener.java:335)
            at org.elasticsearch.action.support.master.TransportMasterNodeAction$AsyncSingleAction$1.handleException(TransportMasterNodeAction.java:269)
```

A similar test failure was reported in #101652 and fixed by #99355, but because the stacktrace is different I report it in a new issue.

https://gradle-enterprise.elastic.co/s/taabgxondrawu/
https://gradle-enterprise.elastic.co/s/vs5aky72bagkk/

**Build scan:**
https://gradle-enterprise.elastic.co/s/vs5aky72bagkk/tests/:server:internalClusterTest/org.elasticsearch.snapshots.ConcurrentSnapshotsIT/testMasterFailoverOnFinalizationLoop
**Reproduction line:**
```
./gradlew ':server:internalClusterTest' --tests "org.elasticsearch.snapshots.ConcurrentSnapshotsIT.testMasterFailoverOnFinalizationLoop" -Dtests.seed=A9C979C6EBC7A9E5 -Dtests.locale=pt-BR -Dtests.timezone=US/Pacific -Druntime.java=21
```

**Applicable branches:**
main, 8.11

**Reproduces locally?:**
Didn't try

**Failure history:**
https://gradle-enterprise.elastic.co/scans/tests?tests.container=org.elasticsearch.snapshots.ConcurrentSnapshotsIT&tests.test=testMasterFailoverOnFinalizationLoop


**Failure excerpt:**
```
java.lang.AssertionError: org.elasticsearch.ElasticsearchException: org.elasticsearch.action.ActionListenerImplementations$RunBeforeActionListener/ChannelActionListener{TaskTransportChannel{task=75}{TcpTransportChannel{req=21}{cluster:admin/snapshot/delete}{Netty4TcpChannel{localAddress=/127.0.0.1:18582, remoteAddress=/127.0.0.1:37788, profile=default}}}}/org.elasticsearch.action.support.master.TransportMasterNodeAction$$Lambda/0x00007ff8b890f628@446be68a

  at __randomizedtesting.SeedInfo.seed([A9C979C6EBC7A9E5]:0)
  at org.elasticsearch.action.ActionListener$4.assertFirstRun(ActionListener.java:324)
  at org.elasticsearch.action.ActionListener$4.onFailure(ActionListener.java:335)
  at org.elasticsearch.action.support.master.TransportMasterNodeAction$AsyncSingleAction$1.handleException(TransportMasterNodeAction.java:269)
  at org.elasticsearch.transport.TransportService$ContextRestoreResponseHandler.handleException(TransportService.java:1421)
  at org.elasticsearch.transport.InboundHandler.doHandleException(InboundHandler.java:475)
  at org.elasticsearch.transport.InboundHandler.handleException(InboundHandler.java:462)
  at org.elasticsearch.transport.InboundHandler.handlerResponseError(InboundHandler.java:453)
  at org.elasticsearch.transport.InboundHandler.executeResponseHandler(InboundHandler.java:145)
  at org.elasticsearch.transport.InboundHandler.messageReceived(InboundHandler.java:122)
  at org.elasticsearch.transport.InboundHandler.inboundMessage(InboundHandler.java:96)
  at org.elasticsearch.transport.TcpTransport.inboundMessage(TcpTransport.java:832)
  at org.elasticsearch.transport.InboundPipeline.forwardFragments(InboundPipeline.java:124)
  at org.elasticsearch.transport.InboundPipeline.doHandleBytes(InboundPipeline.java:96)
  at org.elasticsearch.transport.InboundPipeline.handleBytes(InboundPipeline.java:61)
  at org.elasticsearch.transport.netty4.Netty4MessageInboundHandler.channelRead(Netty4MessageInboundHandler.java:48)
  at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
  at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
  at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
  at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103)
  at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444)
  at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
  at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
  at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
  at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440)
  at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
  at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
  at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
  at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:788)
  at io.netty.channel.nio.NioEventLoop.processSelectedKeysPlain(NioEventLoop.java:689)
  at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:652)
  at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562)
  at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
  at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
  at java.lang.Thread.run(Thread.java:1583)

  Caused by: org.elasticsearch.ElasticsearchException: org.elasticsearch.action.ActionListenerImplementations$RunBeforeActionListener/ChannelActionListener{TaskTransportChannel{task=75}{TcpTransportChannel{req=21}{cluster:admin/snapshot/delete}{Netty4TcpChannel{localAddress=/127.0.0.1:18582, remoteAddress=/127.0.0.1:37788, profile=default}}}}/org.elasticsearch.action.support.master.TransportMasterNodeAction$$Lambda/0x00007ff8b890f628@446be68a

    at org.elasticsearch.action.ActionListener$4.assertFirstRun(ActionListener.java:323)
    at org.elasticsearch.action.ActionListener$4.onFailure(ActionListener.java:335)
    at org.elasticsearch.action.support.master.TransportMasterNodeAction$AsyncSingleAction.lambda$doStart$2(TransportMasterNodeAction.java:233)
    at org.elasticsearch.action.ActionListenerImplementations$DelegatingResponseActionListener.acceptException(ActionListenerImplementations.java:186)
    at org.elasticsearch.action.ActionListenerImplementations.safeAcceptException(ActionListenerImplementations.java:62)
    at org.elasticsearch.action.ActionListenerImplementations$DelegatingResponseActionListener.onFailure(ActionListenerImplementations.java:191)
    at org.elasticsearch.action.ActionListenerImplementations.safeAcceptException(ActionListenerImplementations.java:62)
    at org.elasticsearch.action.ActionListenerImplementations.safeOnFailure(ActionListenerImplementations.java:73)
    at org.elasticsearch.action.DelegatingActionListener.onFailure(DelegatingActionListener.java:27)
    at org.elasticsearch.action.ActionListenerImplementations.safeAcceptException(ActionListenerImplementations.java:62)
    at org.elasticsearch.action.ActionListenerImplementations.safeOnFailure(ActionListenerImplementations.java:73)
    at org.elasticsearch.action.DelegatingActionListener.onFailure(DelegatingActionListener.java:27)
    at org.elasticsearch.action.support.ContextPreservingActionListener.onFailure(ContextPreservingActionListener.java:39)
    at org.elasticsearch.action.ActionListener.onFailure(ActionListener.java:247)
    at org.elasticsearch.snapshots.SnapshotsService.failListenersIgnoringException(SnapshotsService.java:2763)
    at org.elasticsearch.snapshots.SnapshotsService.failAllListenersOnMasterFailOver(SnapshotsService.java:2479)
    at org.elasticsearch.snapshots.SnapshotsService$RemoveSnapshotDeletionAndContinueTask.onFailure(SnapshotsService.java:2532)
    at org.elasticsearch.cluster.service.MasterService$ExecutionResult.notifyFailure(MasterService.java:975)
    at org.elasticsearch.cluster.service.MasterService.executeAndPublishBatch(MasterService.java:223)
    at org.elasticsearch.cluster.service.MasterService$BatchingTaskQueue$Processor.lambda$run$2(MasterService.java:1626)
    at org.elasticsearch.action.ActionListener.run(ActionListener.java:368)
    at org.elasticsearch.cluster.service.MasterService$BatchingTaskQueue$Processor.run(MasterService.java:1623)
    at org.elasticsearch.cluster.service.MasterService$5.lambda$doRun$0(MasterService.java:1237)
    at org.elasticsearch.action.ActionListener.run(ActionListener.java:368)
    at org.elasticsearch.cluster.service.MasterService$5.doRun(MasterService.java:1216)
    at org.elasticsearch.common.util.concurrent.ThreadContext$ContextPreservingAbstractRunnable.doRun(ThreadContext.java:983)
    at org.elasticsearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:26)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
    at java.lang.Thread.run(Thread.java:1583)

```
server/src/internalClusterTest/java/org/elasticsearch/snapshots/ConcurrentSnapshotsIT.java
createNSnapshots(repoName, randomIntBetween(1, 5));
        blockNodeOnAnyFiles(repoName, masterName);
        final int deletes = randomIntBetween(2, 10);
        final List<ActionFuture<AcknowledgedResponse>> deleteResponses = new ArrayList<>(deletes);
        for (int i = 0; i < deletes; ++i) {
            deleteResponses.add(clusterAdmin().prepareDeleteSnapshot(repoName, "*").execute());
        }
        waitForBlock(masterName, repoName);
        awaitNDeletionsInProgress(1);
        for (ActionFuture<AcknowledgedResponse> deleteResponse : deleteResponses) {
            assertFalse(deleteResponse.isDone());
        }
        awaitNDeletionsInProgress(1);
        unblockNode(repoName, masterName);
        for (ActionFuture<AcknowledgedResponse> deleteResponse : deleteResponses) {
            assertAcked(deleteResponse.get());
        }
    }
    public void testMasterFailoverOnFinalizationLoop() throws Exception {
        internalCluster().startMasterOnlyNodes(3);
        final String dataNode = internalCluster().startDataOnlyNode();
        final String repoName = "test-repo";
        createRepository(repoName, "mock");
        createIndexWithContent("index-test");
        final NetworkDisruption networkDisruption = isolateMasterDisruption(NetworkDisruption.DISCONNECT);
        internalCluster().setDisruptionScheme(networkDisruption);
        final List<String> snapshotNames = createNSnapshots(repoName, randomIntBetween(2, 5));
        final String masterName = internalCluster().getMasterName();
        blockMasterFromDeletingIndexNFile(repoName);
        final ActionFuture<CreateSnapshotResponse> snapshotOther = startFullSnapshotFromMasterClient(repoName, "snap-other");