Add optional compression to the OpenTelemetry gRPC requests

modules/opentelemetry/src/main/java/com/spotify/ffwd/opentelemetry/OpenTelemetryOutputPlugin.java

@@ -41,18 +41,21 @@ public class OpenTelemetryOutputPlugin extends OutputPlugin {
 
   private final Map<String, String> headers;
   private final String endpoint;
+  private final String compression;
 
   @JsonCreator
   public OpenTelemetryOutputPlugin(
       @JsonProperty("filter") Optional<Filter> filter,
       @JsonProperty("flushInterval") @Nullable Long flushInterval,
       @JsonProperty("batching") Optional<Batching> batching,
       @JsonProperty("headers") Optional<Map<String, String>> headers,
-      @JsonProperty("endpoint") @Nullable String endpoint
+      @JsonProperty("endpoint") @Nullable String endpoint,
+      @JsonProperty("compression") @Nullable String compression
   ) {
     super(filter, Batching.from(flushInterval, batching));
     this.headers = headers.orElse(new HashMap<>());
     this.endpoint = Objects.requireNonNull(endpoint, "endpoint must be set");
+    this.compression = compression;
   }
 
   @Override
@@ -62,7 +65,7 @@ public Module module(Key<PluginSink> key, String id) {
       protected void configure() {
         bind(Logger.class).toInstance(LoggerFactory.getLogger(id));
         bind(key).toInstance(
-            new OpenTelemetryPluginSink(endpoint, headers));
+            new OpenTelemetryPluginSink(endpoint, headers, compression));
         expose(key);
       }
     };

modules/opentelemetry/src/main/java/com/spotify/ffwd/opentelemetry/OpenTelemetryPluginSink.java

@@ -43,6 +43,7 @@
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
+import javax.annotation.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -55,13 +56,16 @@ public class OpenTelemetryPluginSink implements PluginSink {
   private Map<String, String> headers;
   private ManagedChannel channel;
   private MetricsServiceGrpc.MetricsServiceBlockingStub stub;
+  @Nullable private String compression;
 
   OpenTelemetryPluginSink(
       String endpoint,
-      Map<String, String> headers
+      Map<String, String> headers,
+      @Nullable String compression
   ) {
     this.endpoint = endpoint;
     this.headers = headers;
+    this.compression = compression;
   }
 
 
@@ -122,6 +126,10 @@ public AsyncFuture<Void> start() {
     MetricsServiceGrpc.MetricsServiceBlockingStub stub =
         MetricsServiceGrpc.newBlockingStub(channel);
 
+    if (compression != null) {
+        stub = stub.withCompression(compression);
+    }
+
     Metadata extraHeaders = new Metadata();
     headers.forEach((key, value) -> {
       Metadata.Key<String> header = Metadata.Key.of(key, Metadata.ASCII_STRING_MARSHALLER);