**Build scan:**
https://gradle-enterprise.elastic.co/s/7ogsa7kjmbcp6/tests/:x-pack:plugin:ccr:internalClusterTest/org.elasticsearch.xpack.ccr.AutoFollowIT/testAutoFollowSoftDeletesDisabled
**Reproduction line:**
```
./gradlew ':x-pack:plugin:ccr:internalClusterTest' --tests "org.elasticsearch.xpack.ccr.AutoFollowIT.testAutoFollowSoftDeletesDisabled" -Dtests.seed=7380EBC21C6AE9 -Dtests.locale=be-BY -Dtests.timezone=Antarctica/Rothera -Druntime.java=21
```

**Applicable branches:**
7.17

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

**Failure history:**
https://gradle-enterprise.elastic.co/scans/tests?tests.container=org.elasticsearch.xpack.ccr.AutoFollowIT&tests.test=testAutoFollowSoftDeletesDisabled
**Failure excerpt:**
```
com.carrotsearch.randomizedtesting.UncaughtExceptionError: Captured an uncaught exception in thread: Thread[id=1100, name=elasticsearch[follower2][generic][T#4], state=RUNNABLE, group=TGRP-AutoFollowIT]

  at __randomizedtesting.SeedInfo.seed([7380EBC21C6AE9:9F813802806BFB06]:0)

  Caused by: java.lang.AssertionError: java.lang.AssertionError: listener.onFailure failed

    at __randomizedtesting.SeedInfo.seed([7380EBC21C6AE9]:0)
    at org.elasticsearch.action.ActionListener$Delegating.onFailure(ActionListener.java:71)
    at org.elasticsearch.action.support.ContextPreservingActionListener.onFailure(ContextPreservingActionListener.java:38)
    at org.elasticsearch.action.ActionListener$Delegating.onFailure(ActionListener.java:66)
    at org.elasticsearch.common.util.concurrent.ListenableFuture.notifyListenerDirectly(ListenableFuture.java:115)
    at org.elasticsearch.common.util.concurrent.ListenableFuture.done(ListenableFuture.java:100)
    at org.elasticsearch.common.util.concurrent.BaseFuture.setException(BaseFuture.java:149)
    at org.elasticsearch.common.util.concurrent.ListenableFuture.onFailure(ListenableFuture.java:147)
    at org.elasticsearch.transport.ClusterConnectionManager.failConnectionListener(ClusterConnectionManager.java:372)
    at org.elasticsearch.transport.ClusterConnectionManager.lambda$connectToNodeOrRetry$7(ClusterConnectionManager.java:249)
    at org.elasticsearch.action.ActionListener$1.onFailure(ActionListener.java:144)
    at org.elasticsearch.action.ActionListener$Delegating.onFailure(ActionListener.java:66)
    at org.elasticsearch.action.ActionListener$Delegating.onFailure(ActionListener.java:66)
    at org.elasticsearch.action.support.ThreadedActionListener$2.doRun(ThreadedActionListener.java:110)
    at org.elasticsearch.common.util.concurrent.ThreadContext$ContextPreservingAbstractRunnable.doRun(ThreadContext.java:777)
    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)

    Caused by: java.lang.AssertionError: listener.onFailure failed

      at org.elasticsearch.action.ActionListener$Delegating.onFailure(ActionListener.java:71)
      at org.elasticsearch.action.support.ContextPreservingActionListener.onFailure(ContextPreservingActionListener.java:38)
      at org.elasticsearch.action.ActionListener$Delegating.onFailure(ActionListener.java:66)
      at org.elasticsearch.common.util.concurrent.ListenableFuture.notifyListenerDirectly(ListenableFuture.java:115)
      at org.elasticsearch.common.util.concurrent.ListenableFuture.done(ListenableFuture.java:100)
      at org.elasticsearch.common.util.concurrent.BaseFuture.setException(BaseFuture.java:149)
      at org.elasticsearch.common.util.concurrent.ListenableFuture.onFailure(ListenableFuture.java:147)
      at org.elasticsearch.transport.ClusterConnectionManager.failConnectionListener(ClusterConnectionManager.java:372)
      at org.elasticsearch.transport.ClusterConnectionManager.lambda$connectToNodeOrRetry$7(ClusterConnectionManager.java:249)
      at org.elasticsearch.action.ActionListener$1.onFailure(ActionListener.java:144)
      at org.elasticsearch.action.ActionListener$Delegating.onFailure(ActionListener.java:66)
      at org.elasticsearch.action.ActionListener$Delegating.onFailure(ActionListener.java:66)
      at org.elasticsearch.action.support.ThreadedActionListener$2.doRun(ThreadedActionListener.java:110)
      at org.elasticsearch.common.util.concurrent.ThreadContext$ContextPreservingAbstractRunnable.doRun(ThreadContext.java:777)
      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)

      Caused by: java.lang.IllegalStateException: transport has been stopped

        at org.elasticsearch.transport.TcpTransport.ensureOpen(TcpTransport.java:899)
        at org.elasticsearch.transport.TcpTransport.openConnection(TcpTransport.java:316)
        at org.elasticsearch.test.transport.StubbableTransport.openConnection(StubbableTransport.java:164)
        at org.elasticsearch.transport.ClusterConnectionManager.internalOpenConnection(ClusterConnectionManager.java:348)
        at org.elasticsearch.transport.ClusterConnectionManager.connectToNodeOrRetry(ClusterConnectionManager.java:199)
        at org.elasticsearch.transport.ClusterConnectionManager.connectToNode(ClusterConnectionManager.java:98)
        at org.elasticsearch.transport.RemoteConnectionManager.connectToRemoteClusterNode(RemoteConnectionManager.java:64)
        at org.elasticsearch.transport.SniffConnectionStrategy$SniffClusterStateResponseHandler.handleNodes(SniffConnectionStrategy.java:450)
        at org.elasticsearch.transport.SniffConnectionStrategy$SniffClusterStateResponseHandler.access$300(SniffConnectionStrategy.java:418)
        at org.elasticsearch.transport.SniffConnectionStrategy$SniffClusterStateResponseHandler$1.onFailure(SniffConnectionStrategy.java:472)
        at org.elasticsearch.action.ActionListener$Delegating.onFailure(ActionListener.java:66)
        at org.elasticsearch.action.support.ContextPreservingActionListener.onFailure(ContextPreservingActionListener.java:38)
        at org.elasticsearch.action.ActionListener$Delegating.onFailure(ActionListener.java:66)
        at org.elasticsearch.common.util.concurrent.ListenableFuture.notifyListenerDirectly(ListenableFuture.java:115)
        at org.elasticsearch.common.util.concurrent.ListenableFuture.done(ListenableFuture.java:100)
        at org.elasticsearch.common.util.concurrent.BaseFuture.setException(BaseFuture.java:149)
        at org.elasticsearch.common.util.concurrent.ListenableFuture.onFailure(ListenableFuture.java:147)
        at org.elasticsearch.transport.ClusterConnectionManager.failConnectionListener(ClusterConnectionManager.java:372)
        at org.elasticsearch.transport.ClusterConnectionManager.lambda$connectToNodeOrRetry$7(ClusterConnectionManager.java:249)
        at org.elasticsearch.action.ActionListener$1.onFailure(ActionListener.java:144)
        at org.elasticsearch.action.ActionListener$Delegating.onFailure(ActionListener.java:66)
        at org.elasticsearch.action.ActionListener$Delegating.onFailure(ActionListener.java:66)
        at org.elasticsearch.action.support.ThreadedActionListener$2.doRun(ThreadedActionListener.java:110)
        at org.elasticsearch.common.util.concurrent.ThreadContext$ContextPreservingAbstractRunnable.doRun(ThreadContext.java:777)
        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)

```
test/framework/src/main/java/org/elasticsearch/transport/AbstractSimpleTransportTestCase.java
}
                    }
                }
                @Override
                public void onAfter() {
                    done.countDown();
                }
            });
        }
        for (int i = 0; i < halfSenders; i++) {
            final int sender = i;
            threadPool.executor(ThreadPool.Names.GENERIC).execute(new AbstractRunnable() {
                @Override
                public void onFailure(Exception e) {
                    logger.error("unexpected error", e);
                    sendingErrors.add(e);
                }
                @Override
                protected void doRun() throws Exception {
                    go.await();
                    for (int iter = 0; iter < 10; iter++) {
                        PlainActionFuture<TestResponse> listener = new PlainActionFuture<>();
                        final String info = sender + "_" + iter;
                        final DiscoveryNode node = nodeB; 
                        try {
                            serviceA.sendRequest(
                                node,
                                "internal:test",
                                new TestRequest(info),
                                new ActionListenerResponseHandler<>(listener, TestResponse::new, TransportResponseHandler.TRANSPORT_WORKER)