Proxy HTTP request headers from VS Code (#99)

* Proxy HTTP request headers from VS Code

* fix import order

* use MultiMap header API to remove headers instead of reimplementing

* Fix GZIP decoding issue in RestAssured

* Exclude more headers

src/main/java/io/confluent/idesidecar/restapi/proxy/clusters/strategy/ClusterStrategy.java

@@ -4,7 +4,9 @@
 import io.confluent.idesidecar.restapi.util.UriUtil;
 import io.vertx.core.MultiMap;
 import io.vertx.core.http.HttpHeaders;
+import java.util.List;
 import java.util.regex.Pattern;
+import org.eclipse.microprofile.config.inject.ConfigProperty;
 
 /**
  * Holds default implementations for processing constructing the proxy URI, headers before
@@ -18,8 +20,14 @@ public abstract class ClusterStrategy {
 
   static UriUtil uriUtil = new UriUtil();
 
+  @ConfigProperty(name = "ide-sidecar.cluster-proxy.http-header-exclusions")
+  List<String> httpHeaderExclusions;
+
   public MultiMap constructProxyHeaders(ClusterProxyContext context) {
-    return HttpHeaders.headers();
+    var headers = HttpHeaders.headers();
+    headers.addAll(context.getRequestHeaders());
+    httpHeaderExclusions.forEach(headers::remove);
+    return headers;
   }
 
   public String constructProxyUri(String requestUri, String clusterUri) {

src/main/java/io/confluent/idesidecar/restapi/proxy/clusters/strategy/ConfluentCloudKafkaClusterStrategy.java

@@ -3,7 +3,6 @@
 import io.confluent.idesidecar.restapi.connections.CCloudConnectionState;
 import io.confluent.idesidecar.restapi.proxy.clusters.ClusterProxyContext;
 import io.vertx.core.MultiMap;
-import io.vertx.core.http.HttpHeaders;
 import jakarta.enterprise.context.ApplicationScoped;
 
 /**
@@ -15,7 +14,7 @@ public class ConfluentCloudKafkaClusterStrategy extends ClusterStrategy {
 
   @Override
   public MultiMap constructProxyHeaders(ClusterProxyContext context) {
-    var headers = HttpHeaders.headers();
+    var headers = super.constructProxyHeaders(context);
     if (context.getConnectionState() instanceof CCloudConnectionState cCloudConnectionState) {
       cCloudConnectionState.getOauthContext()
           .getDataPlaneAuthenticationHeaders()

src/main/java/io/confluent/idesidecar/restapi/proxy/clusters/strategy/ConfluentCloudSchemaRegistryClusterStrategy.java

@@ -3,7 +3,6 @@
 import io.confluent.idesidecar.restapi.connections.CCloudConnectionState;
 import io.confluent.idesidecar.restapi.proxy.clusters.ClusterProxyContext;
 import io.vertx.core.MultiMap;
-import io.vertx.core.http.HttpHeaders;
 import jakarta.enterprise.context.ApplicationScoped;
 
 /**
@@ -16,7 +15,7 @@ public class ConfluentCloudSchemaRegistryClusterStrategy extends ClusterStrategy
 
   @Override
   public MultiMap constructProxyHeaders(ClusterProxyContext context) {
-    var headers = HttpHeaders.headers();
+    var headers = super.constructProxyHeaders(context);
     if (context.getConnectionState() instanceof CCloudConnectionState cCloudConnectionState) {
       cCloudConnectionState.getOauthContext()
           .getDataPlaneAuthenticationHeaders()

src/main/java/io/confluent/idesidecar/restapi/proxy/clusters/strategy/ConfluentLocalKafkaClusterStrategy.java

@@ -31,8 +31,8 @@ public class ConfluentLocalKafkaClusterStrategy extends ClusterStrategy {
    */
   @Override
   public MultiMap constructProxyHeaders(ClusterProxyContext context) {
-    return HttpHeaders
-        .headers()
+    var headers = super.constructProxyHeaders(context);
+    return headers
         .add(CONNECTION_ID_HEADER, context.getConnectionId())
         .add(HttpHeaders.AUTHORIZATION, "Bearer %s".formatted(accessToken.getToken()));
   }

src/main/resources/application.yml

@@ -154,6 +154,18 @@ ide-sidecar:
     "auto.register.schemas": "false"
     # Do not try to fetch the latest version of the schema from the registry
     "use.latest.version": "false"
+  cluster-proxy:
+    # Exclude these HTTP headers when proxying requests to the remote cluster (Kafka or Schema Registry)
+    # It shouldn't really hurt if we _don't_ exclude these headers, but it's just cleaner to do so.
+    http-header-exclusions:
+      - x-connection-id
+      - x-cluster-id
+      - Authorization
+      - Host
+      - Connection
+      # This may contradict UTF-8 decoding done in the proxy response processing
+      # https://github.com/confluentinc/ide-sidecar/issues/102
+      - Accept-Encoding
 
 quarkus:
   application:

src/test/java/io/confluent/idesidecar/restapi/kafkarest/RecordsV3ApiImplIT.java

@@ -77,6 +77,7 @@ class SadPath {
     @Test
     void shouldThrowNotFoundWhenClusterDoesNotExist() {
       givenConnectionId()
+          .header("Content-Type", "application/json")
           .body(createProduceRequest(
               null, "key", null, "value", null
           ))

src/test/java/io/confluent/idesidecar/restapi/resources/ClusterRestProxyResourceTest.java

@@ -216,6 +216,9 @@ void testKafkaRestProxyAgainstCCloud() throws Throwable {
                     .withStatus(200)
                     .withHeader("Content-Type", "application/json")
                     .withHeader("x-ccloud-specific-header", "fake-value")
+                    // Explicitly disable gzip since RestAssured sends the
+                    // Accept-Encoding: gzip header by default
+                    .withGzipDisabled(true)
                     .withBody(
                         new String(Objects.requireNonNull(
                             Thread
@@ -275,16 +278,21 @@ void testSchemaRegistryRestProxyAgainstCCloud() throws Throwable {
 
     // Given we have a fake CCloud Schema Registry server endpoint for list schemas
     wireMock.register(
-        WireMock.get("/schemas")
+        WireMock
+            .get("/schemas")
             .withHeader("Authorization",
                 new EqualToPattern("Bearer %s".formatted(dataPlaneToken.token()))
             )
             .withHeader("target-sr-cluster", new EqualToPattern(srClusterId))
+            .withHeader("x-non-sidecar-specific-header", new EqualToPattern("dummy"))
             .willReturn(
                 WireMock.aResponse()
                     .withStatus(200)
                     .withHeader("Content-Type", "application/json")
                     .withHeader("x-ccloud-specific-header", "fake-value")
+                    // Explicitly disable gzip since RestAssured sends the
+                    // Accept-Encoding: gzip header by default
+                    .withGzipDisabled(true)
                     .withBody(
                         new String(Objects.requireNonNull(
                             Thread
@@ -301,7 +309,10 @@ void testSchemaRegistryRestProxyAgainstCCloud() throws Throwable {
         .when()
         .headers(Map.of(
             "x-connection-id", CONNECTION_ID,
-            "x-cluster-id", srClusterId
+            "x-cluster-id", srClusterId,
+            // Assert that any headers sent to the proxy are passed through
+            // to the target server
+             "x-non-sidecar-specific-header", "dummy"
         ))
         .get("/schemas")
         .then();

src/test/java/io/confluent/idesidecar/restapi/util/ConfluentLocalTestBed.java

@@ -11,6 +11,8 @@
 import io.confluent.idesidecar.restapi.testutil.NoAccessFilterProfile;
 import io.confluent.kafka.schemaregistry.client.rest.entities.Schema;
 import io.quarkus.test.junit.TestProfile;
+import io.restassured.RestAssured;
+import io.restassured.config.DecoderConfig;
 import io.restassured.http.ContentType;
 import io.restassured.response.ValidatableResponse;
 import io.restassured.specification.RequestSpecification;
@@ -218,6 +220,13 @@ protected static RequestSpecification givenConnectionId() {
 
   protected static RequestSpecification givenConnectionId(String connectionId) {
     return given()
+        .config(
+            // https://stackoverflow.com/a/67876342 saves the day
+            // Not specifying this will lead to a `java.util.zip.ZipException: Not in GZIP format` error
+            // even though the response is not gzipped, because RestAssured tries to decode it as such
+            // by default. This is a workaround to disable the default decoders.
+            RestAssured.config().decoderConfig(DecoderConfig.decoderConfig().noContentDecoders())
+        )
         .header("X-connection-id", connectionId);
   }
 
@@ -356,6 +365,7 @@ public Schema createSchema(String subject, String schemaType, String schema) {
     var srCluster = getSchemaRegistryCluster();
     var createSchemaVersionResp = givenConnectionId()
         .headers(
+             "Content-Type", "application/json",
             "X-cluster-id", srCluster.id()
         )
         .body(Map.of(