Ensure indexing_data CCR requests are compressed (#79413)

Currently, CCR requests are not compressed if they are indexing data.
This is because the BulkShardOperationsRequest does not
implement the indexing data interface.

Additionally, the fact that the request can be wrapped in proxy request
also requires special logic. This commit implements the same logic that
we do for shard indexing actions to ensure that the data is compressed
if the underlying request is indexing data.

Author: Tim-Brooks

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

@@ -100,7 +100,7 @@ public void handleException(TransportException exp) {
         }
     }
 
-    static class ProxyRequest<T extends TransportRequest> extends TransportRequest {
+    static class ProxyRequest<T extends TransportRequest> extends TransportRequest implements RawIndexingDataTransportRequest {
         final T wrapped;
         final DiscoveryNode targetNode;
 
@@ -122,6 +122,14 @@ public void writeTo(StreamOutput out) throws IOException {
             targetNode.writeTo(out);
             wrapped.writeTo(out);
         }
+
+        @Override
+        public boolean isRawIndexingData() {
+            if (wrapped instanceof RawIndexingDataTransportRequest) {
+                return ((RawIndexingDataTransportRequest) wrapped).isRawIndexingData();
+            }
+            return false;
+        }
     }
 
     private static class CancellableProxyRequest<T extends TransportRequest> extends ProxyRequest<T> {

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/bulk/BulkShardOperationsRequest.java

@@ -11,11 +11,13 @@
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.index.translog.Translog;
+import org.elasticsearch.transport.RawIndexingDataTransportRequest;
 
 import java.io.IOException;
 import java.util.List;
 
-public final class BulkShardOperationsRequest extends ReplicatedWriteRequest<BulkShardOperationsRequest> {
+public final class BulkShardOperationsRequest extends ReplicatedWriteRequest<BulkShardOperationsRequest>
+    implements RawIndexingDataTransportRequest {
 
     private final String historyUUID;
     private final List<Translog.Operation> operations;