diff --git a/ksqldb-rest-client/src/main/java/io/confluent/ksql/rest/client/KsqlRestClient.java b/ksqldb-rest-client/src/main/java/io/confluent/ksql/rest/client/KsqlRestClient.java
index 363a69e520da..c0b4c53c37f0 100644
--- a/ksqldb-rest-client/src/main/java/io/confluent/ksql/rest/client/KsqlRestClient.java
+++ b/ksqldb-rest-client/src/main/java/io/confluent/ksql/rest/client/KsqlRestClient.java
@@ -66,6 +66,33 @@ public final class KsqlRestClient implements Closeable {
   private List<URI> serverAddresses;
   private boolean isCCloudServer;
 
+
+
+  /**
+   * @param serverAddress the address of the KSQL server to connect to.
+   * @param localProps initial set of local properties.
+   * @param clientProps properties used to build the client.
+   * @param creds optional credentials
+   */
+  @Deprecated
+  public static KsqlRestClient create(
+      final String serverAddress,
+      final Map<String, ?> localProps,
+      final Map<String, String> clientProps,
+      final Optional<BasicCredentials> creds
+  ) {
+    return create(
+        serverAddress,
+        localProps,
+        clientProps,
+        creds,
+        Optional.empty(),
+        (cprops, credz, lprops) -> new KsqlClient(cprops, credz, lprops,
+            new HttpClientOptions(),
+            Optional.of(new HttpClientOptions().setProtocolVersion(HttpVersion.HTTP_2)))
+    );
+  }
+
   /**
    * @param serverAddress the address of the KSQL server to connect to.
    * @param localProps initial set of local properties.
diff --git a/ksqldb-rest-client/src/test/java/io/confluent/ksql/rest/client/KsqlRestClientTest.java b/ksqldb-rest-client/src/test/java/io/confluent/ksql/rest/client/KsqlRestClientTest.java
index 4b12658557da..67bfe4730b69 100644
--- a/ksqldb-rest-client/src/test/java/io/confluent/ksql/rest/client/KsqlRestClientTest.java
+++ b/ksqldb-rest-client/src/test/java/io/confluent/ksql/rest/client/KsqlRestClientTest.java
@@ -19,6 +19,7 @@
 import static io.confluent.ksql.rest.client.KsqlRestClient.CCLOUD_CONNECT_USERNAME_HEADER;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
 import static org.junit.Assert.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
@@ -222,6 +223,18 @@ public void shouldNotIncludeAdditionalHeadersForOnPremNonConnectorRequest() thro
     verify(target).postKsqlRequest("some ksql;", Collections.emptyMap(), Optional.of(0L));
   }
 
+  @Test
+  public void shouldAllowCallingCreateWithoutNeedingACCloudApiKey() {
+    // This is a backwards compatibility check
+
+    final KsqlRestClient client = KsqlRestClient.create(SOME_SERVER_ADDRESS, LOCAL_PROPS, CLIENT_PROPS, Optional.empty());
+    assertThat(client, is(instanceOf(KsqlRestClient.class)));
+
+    // Also with new signature
+    final KsqlRestClient ccloudClient = KsqlRestClient.create(SOME_SERVER_ADDRESS, LOCAL_PROPS, CLIENT_PROPS, Optional.empty(), Optional.empty());
+    assertThat(ccloudClient, is(instanceOf(KsqlRestClient.class)));
+  }
+
   private KsqlRestClient clientWithServerAddresses(final String serverAddresses) {
     return clientWithServerAddresses(serverAddresses, Optional.empty());
   }