fix: make api client recognize ddl warnings better (#9341)

ksqldb-api-client/src/main/java/io/confluent/ksql/api/client/impl/DdlDmlResponseHandlers.java

@@ -73,8 +73,8 @@ private static boolean isCommandStatusEntity(final JsonObject ksqlEntity) {
   }
 
   private static boolean isIfNotExistsWarning(final JsonObject ksqlEntity) {
-    return ksqlEntity.getString("statementText") != null
-        && ksqlEntity.getString("statementText").contains("IF NOT EXISTS")
+    return ksqlEntity.getString("message") != null
+        && ksqlEntity.getString("message").startsWith("Cannot add")
         && ksqlEntity.getString("@type") != null
         && ksqlEntity.getString("@type").equals("warning_entity");
   }

ksqldb-api-client/src/test/java/io/confluent/ksql/api/client/ClientTest.java

@@ -30,6 +30,7 @@
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.nullValue;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertThrows;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
@@ -1191,6 +1192,40 @@ public void shouldFailToDropConnectorViaExecuteStatement() {
         containsString("Use the dropConnector() method instead"));
   }
 
+  @Test
+  public void shouldHandleWarningsForDdlStatements()
+      throws ExecutionException, InterruptedException {
+    // Given
+    final WarningEntity entity = new WarningEntity("create stream if not exists;", "Cannot add stream foo: A stream with the same name already exists.");
+    testEndpoints.setKsqlEndpointResponse(Collections.singletonList(entity));
+
+    // When
+    final ExecuteStatementResult result = javaClient.executeStatement("create stream if not exists;").get();
+
+    // Then
+    assertFalse(result.queryId().isPresent());
+  }
+
+  @Test
+  public void shouldRejectWarningsFromNonDdlStatementsViaExecuteStatement() {
+    // Given
+    final WarningEntity entity = new WarningEntity("drop connector if exits;", "Connector does not exist.");
+    testEndpoints.setKsqlEndpointResponse(Collections.singletonList(entity));
+
+    // When
+    final Exception e = assertThrows(
+        ExecutionException.class, // thrown from .get() when the future completes exceptionally
+        () -> javaClient.executeStatement("drop connector if exits;").get()
+    );
+
+    // Then
+    assertThat(e.getCause(), instanceOf(KsqlClientException.class));
+    assertThat(e.getCause().getMessage(), containsString(EXECUTE_STATEMENT_REQUEST_ACCEPTED_DOC));
+    assertThat(e.getCause().getMessage(), containsString(EXECUTE_STATEMENT_USAGE_DOC));
+    assertThat(e.getCause().getMessage(),
+        containsString("Use the dropConnector() method instead"));
+  }
+
   @Test
   public void shouldListStreams() throws Exception {
     // Given

ksqldb-api-client/src/test/java/io/confluent/ksql/api/client/integration/ClientIntegrationTest.java

@@ -34,6 +34,7 @@
 import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.nullValue;
 import static org.hamcrest.Matchers.startsWith;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.fail;
 
@@ -829,6 +830,34 @@ public void shouldHandleErrorResponseFromExecuteStatement() {
     assertThat(e.getCause().getMessage(), containsString("Error code: 40001"));
   }
 
+  @Test
+  public void shouldHandleWarningResponseFromExecuteStatement()
+      throws ExecutionException, InterruptedException {
+    // Given:
+    client.executeStatement("create table if not exists tasks (taskId varchar primary key) with (kafka_topic='tasks', value_format='json', partitions=1);").get();
+
+    // When:
+    final ExecuteStatementResult result =
+        client.executeStatement("create table if not exists tasks (taskId varchar primary key) with (kafka_topic='tasks', value_format='json', partitions=1);").get();
+
+    // Then
+    assertFalse(result.queryId().isPresent());
+  }
+
+  @Test
+  public void shouldRejectWarningsFromConnectorRequestsInExecuteStatement() throws Exception {
+    // When
+    final Exception e = assertThrows(
+        ExecutionException.class, // thrown from .get() when the future completes exceptionally
+        () -> client.executeStatement("DROP CONNECTOR IF EXISTS foo;").get()
+    );
+
+    // Then
+    assertThat(e.getCause(), instanceOf(KsqlClientException.class));
+    assertThat(e.getCause().getMessage(), containsString("The ksqlDB server accepted the statement issued via executeStatement(), but the response received is of an unexpected format."));
+    assertThat(e.getCause().getMessage(), containsString("Use the dropConnector() method instead."));
+  }
+
   @Test
   public void shouldRejectMultipleRequestsFromExecuteStatement() {
     // When