Fix memory/breaker leaks for outbound responses (#76474) (#76536)

Outbound responses would not get the expected `decRef`, resulting in
memory and/or circuit breaker leaks. In particular, the
`GetCcrRestoreFileChunkResponse` expects this, causing a leak when
a follower bootstraps.

Relates #65921

server/src/main/java/org/elasticsearch/transport/OutboundHandler.java

@@ -114,7 +114,13 @@ void sendResponse(final Version nodeVersion, final Set<String> features, final T
         }
         OutboundMessage.Response message = new OutboundMessage.Response(threadPool.getThreadContext(), features, response, version,
             requestId, isHandshake, compressionScheme);
-        ActionListener<Void> listener = ActionListener.wrap(() -> messageListener.onResponseSent(requestId, action, response));
+        ActionListener<Void> listener = ActionListener.wrap(() -> {
+                try {
+                    messageListener.onResponseSent(requestId, action, response);
+                } finally {
+                    response.decRef();
+                }
+        });
         sendMessage(channel, message, listener);
     }
 

server/src/main/java/org/elasticsearch/transport/TransportActionProxy.java

@@ -83,6 +83,7 @@ public T read(StreamInput in) throws IOException {
         @Override
         public void handleResponse(T response) {
             try {
+                response.incRef();
                 channel.sendResponse(response);
             } catch (IOException e) {
                 throw new UncheckedIOException(e);

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/CcrIntegTestCase.java

@@ -260,6 +260,9 @@ public void afterTest() throws Exception {
             clusterGroup.leaderCluster.wipe(Collections.emptySet());
             clusterGroup.followerCluster.wipe(Collections.emptySet());
         }
+
+        clusterGroup.leaderCluster.assertAfterTest();
+        clusterGroup.followerCluster.assertAfterTest();
     }
 
     private NodeConfigurationSource createNodeConfigurationSource(final String leaderSeedAddress, final boolean leaderCluster) {