test: add RestQueryTranslationTest

ksql-cli/src/test/java/io/confluent/ksql/cli/CliTest.java

@@ -56,10 +56,10 @@
 import io.confluent.ksql.rest.server.TestKsqlRestApp;
 import io.confluent.ksql.rest.server.computation.CommandId;
 import io.confluent.ksql.rest.server.resources.Errors;
+import io.confluent.ksql.rest.server.resources.RootDocument;
 import io.confluent.ksql.schema.ksql.LogicalSchema;
 import io.confluent.ksql.schema.ksql.PhysicalSchema;
 import io.confluent.ksql.serde.SerdeOption;
-import io.confluent.ksql.rest.server.resources.RootDocument;
 import io.confluent.ksql.test.util.EmbeddedSingleNodeKafkaCluster;
 import io.confluent.ksql.test.util.KsqlIdentifierTestUtil;
 import io.confluent.ksql.util.KsqlConfig;
@@ -88,6 +88,8 @@
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.data.SchemaBuilder;
 import org.apache.kafka.streams.StreamsConfig;
+import org.eclipse.jetty.http.HttpStatus;
+import org.eclipse.jetty.http.HttpStatus.Code;
 import org.hamcrest.Matcher;
 import org.hamcrest.Matchers;
 import org.junit.After;
@@ -702,9 +704,11 @@ public void shouldPrintErrorOnUnsupportedAPI() throws Exception {
     final KsqlRestClient mockRestClient = givenMockRestClient();
     when(mockRestClient.makeRootRequest()).thenReturn(
         RestResponse.erroneous(
+            Code.NOT_ACCEPTABLE,
             new KsqlErrorMessage(
                 Errors.toErrorCode(NOT_ACCEPTABLE.getStatusCode()),
-                "Minimum supported client version: 1.0")));
+                "Minimum supported client version: 1.0")
+        ));
 
     new Cli(1L, 1L, mockRestClient, console)
         .runInteractively();
@@ -867,8 +871,10 @@ public void shouldUpdateCommandSequenceNumber() throws Exception {
     final KsqlRestClient mockRestClient = givenMockRestClient();
     final CommandStatusEntity stubEntity = stubCommandStatusEntityWithSeqNum(12L);
     when(mockRestClient.makeKsqlRequest(anyString(), anyLong()))
-        .thenReturn(RestResponse.successful(new KsqlEntityList(
-            Collections.singletonList(stubEntity))));
+        .thenReturn(RestResponse.successful(
+            HttpStatus.Code.OK,
+            new KsqlEntityList(Collections.singletonList(stubEntity))
+        ));
 
     // When:
     localCli.handleLine(statementText);
@@ -885,8 +891,10 @@ public void shouldUpdateCommandSequenceNumberOnMultipleCommandStatusEntities() t
     final CommandStatusEntity firstEntity = stubCommandStatusEntityWithSeqNum(12L);
     final CommandStatusEntity secondEntity = stubCommandStatusEntityWithSeqNum(14L);
     when(mockRestClient.makeKsqlRequest(anyString(), anyLong()))
-        .thenReturn(RestResponse.successful(new KsqlEntityList(
-            ImmutableList.of(firstEntity, secondEntity))));
+        .thenReturn(RestResponse.successful(
+            HttpStatus.Code.OK,
+            new KsqlEntityList(ImmutableList.of(firstEntity, secondEntity))
+        ));
 
     // When:
     localCli.handleLine(statementText);
@@ -901,7 +909,7 @@ public void shouldNotUpdateCommandSequenceNumberIfNoCommandStatusEntities() thro
     final String statementText = "create stream foo;";
     final KsqlRestClient mockRestClient = givenMockRestClient();
     when(mockRestClient.makeKsqlRequest(anyString(), anyLong()))
-        .thenReturn(RestResponse.successful(new KsqlEntityList()));
+        .thenReturn(RestResponse.successful(HttpStatus.Code.OK, new KsqlEntityList()));
 
     // When:
     localCli.handleLine(statementText);
@@ -916,7 +924,7 @@ public void shouldIssueRequestWithoutCommandSequenceNumberIfRequestPipeliningOn(
     final String statementText = "create stream foo;";
     final KsqlRestClient mockRestClient = givenMockRestClient();
     when(mockRestClient.makeKsqlRequest(anyString(), eq(null)))
-        .thenReturn(RestResponse.successful(new KsqlEntityList()));
+        .thenReturn(RestResponse.successful(HttpStatus.Code.OK, new KsqlEntityList()));
 
     givenRequestPipelining("ON");
 
@@ -934,8 +942,10 @@ public void shouldUpdateCommandSequenceNumberEvenIfRequestPipeliningOn() throws
     final KsqlRestClient mockRestClient = givenMockRestClient();
     final CommandStatusEntity stubEntity = stubCommandStatusEntityWithSeqNum(12L);
     when(mockRestClient.makeKsqlRequest(anyString(), eq(null)))
-        .thenReturn(RestResponse.successful(new KsqlEntityList(
-            Collections.singletonList(stubEntity))));
+        .thenReturn(RestResponse.successful(
+            HttpStatus.Code.OK,
+            new KsqlEntityList(Collections.singletonList(stubEntity))
+        ));
 
     givenRequestPipelining("ON");
 
@@ -975,7 +985,7 @@ public void shouldResetStateWhenServerChanges() throws Exception {
     givenCommandSequenceNumber(mockRestClient, 5L);
     givenRequestPipelining("ON");
     when(mockRestClient.makeRootRequest()).thenReturn(
-        RestResponse.successful(new RootDocument()));
+        RestResponse.successful(HttpStatus.Code.OK, new RootDocument()));
 
     // When:
     runCliSpecificCommand("server foo");
@@ -1000,8 +1010,11 @@ private void givenRunInteractivelyWillExit() {
   private KsqlRestClient givenMockRestClient() throws Exception {
     final KsqlRestClient mockRestClient = mock(KsqlRestClient.class);
 
-    when(mockRestClient.getServerInfo()).thenReturn(
-        RestResponse.of(new ServerInfo("1.x", "testClusterId", "testServiceId")));
+    when(mockRestClient.getServerInfo()).thenReturn(RestResponse.successful(
+        HttpStatus.Code.OK,
+        new ServerInfo("1.x", "testClusterId", "testServiceId")
+    ));
+
     when(mockRestClient.getServerAddress()).thenReturn(new URI("http://someserver:8008"));
 
     localCli = new Cli(
@@ -1022,9 +1035,11 @@ private static CommandStatusEntity stubCommandStatusEntityWithSeqNum(final long
   private void givenCommandSequenceNumber(
       final KsqlRestClient mockRestClient, final long seqNum) throws Exception {
     final CommandStatusEntity stubEntity = stubCommandStatusEntityWithSeqNum(seqNum);
-    when(mockRestClient.makeKsqlRequest(anyString(), anyLong()))
-        .thenReturn(RestResponse.successful(new KsqlEntityList(
-            Collections.singletonList(stubEntity))));
+    when(mockRestClient.makeKsqlRequest(anyString(), anyLong())).thenReturn(
+        RestResponse.successful(
+            HttpStatus.Code.OK,
+            new KsqlEntityList(Collections.singletonList(stubEntity))
+        ));
     localCli.handleLine("create stream foo;");
   }
 
@@ -1034,7 +1049,7 @@ private void assertLastCommandSequenceNumber(
     reset(mockRestClient);
     final String statementText = "list streams;";
     when(mockRestClient.makeKsqlRequest(anyString(), anyLong()))
-        .thenReturn(RestResponse.successful(new KsqlEntityList()));
+        .thenReturn(RestResponse.successful(HttpStatus.Code.OK, new KsqlEntityList()));
 
     // When:
     localCli.handleLine(statementText);

ksql-cli/src/test/java/io/confluent/ksql/cli/console/cmd/RemoteServerSpecificCommandTest.java

@@ -28,15 +28,14 @@
 import io.confluent.ksql.rest.client.KsqlRestClient;
 import io.confluent.ksql.rest.client.RestResponse;
 import io.confluent.ksql.rest.client.exception.KsqlRestClientException;
-import io.confluent.ksql.rest.entity.ServerInfo;
-import io.confluent.ksql.rest.server.resources.Errors;
 import io.confluent.ksql.rest.server.resources.RootDocument;
 import io.confluent.ksql.util.Event;
 import java.io.PrintWriter;
 import java.io.StringWriter;
 import java.net.URI;
 import javax.net.ssl.SSLException;
 import javax.ws.rs.ProcessingException;
+import org.eclipse.jetty.http.HttpStatus.Code;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -65,7 +64,7 @@ public void setUp() throws Exception {
     terminal = new PrintWriter(out);
     command = RemoteServerSpecificCommand.create(restClient, resetCliForNewServer);
 
-    when(restClient.makeRootRequest()).thenReturn(RestResponse.successful(ROOT_DOCUMENT));
+    when(restClient.makeRootRequest()).thenReturn(RestResponse.successful(Code.OK, ROOT_DOCUMENT));
     when(restClient.getServerAddress()).thenReturn(new URI(INITIAL_SERVER_ADDRESS));
   }
 
@@ -125,7 +124,7 @@ public void shouldOutputNewServerDetails() {
   public void shouldPrintErrorOnErrorResponseFromRestClient() {
     // Given:
     when(restClient.makeRootRequest()).thenReturn(RestResponse.erroneous(
-        Errors.ERROR_CODE_SERVER_ERROR, "it is broken"));
+        Code.INTERNAL_SERVER_ERROR, "it is broken"));
 
     // When:
     command.execute(ImmutableList.of(VALID_SERVER_ADDRESS), terminal);

ksql-engine/src/test/java/io/confluent/ksql/integration/IntegrationTestHarness.java

@@ -477,26 +477,6 @@ public void waitForTopicsToBePresent(final String... topicNames) throws Exceptio
         "topics not all present after 30 seconds. topics: " + Arrays.toString(topicNames));
   }
 
-  /**
-   * Wait for topics with names {@code topicNames} to not exist in Kafka.
-   *
-   * @param topicNames the names of the topics to await absence for.
-   */
-  public void waitForTopicsToBeAbsent(final String... topicNames) throws Exception {
-    TestUtils.waitForCondition(
-        () -> {
-          try {
-            final KafkaTopicClient topicClient = serviceContext.get().getTopicClient();
-            return Arrays.stream(topicNames)
-                .noneMatch(topicClient::isTopicExists);
-          } catch (Exception e) {
-            throw new RuntimeException("could not get subjects");
-          }
-        },
-        30_000,
-        "topics not all absent after 30 seconds. topics: " + Arrays.toString(topicNames));
-  }
-
   /**
    * Wait for a subject with name {@code subjectName} to exist in Schema Registry.
    *

ksql-functional-tests/pom.xml

@@ -74,6 +74,21 @@
       <artifactId>kafka-streams-test-utils</artifactId>
     </dependency>
 
+    <dependency>
+      <groupId>io.confluent.ksql</groupId>
+      <artifactId>ksql-rest-app</artifactId>
+      <version>${project.version}</version>
+      <scope>test</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>io.confluent.ksql</groupId>
+      <artifactId>ksql-rest-app</artifactId>
+      <version>${project.version}</version>
+      <type>test-jar</type>
+      <scope>test</scope>
+    </dependency>
+
     <dependency>
       <groupId>io.confluent.ksql</groupId>
       <artifactId>ksql-test-util</artifactId>

ksql-functional-tests/src/main/java/io/confluent/ksql/engine/FakeInsertValuesExecutor.java

@@ -26,6 +26,7 @@
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.Objects;
+import java.util.Optional;
 import org.apache.kafka.clients.producer.ProducerRecord;
 
 public final class FakeInsertValuesExecutor  {
@@ -59,7 +60,7 @@ void sendRecord(final ProducerRecord<byte[], byte[]> record) {
                   fakeKafkaService.getTopic(record.topic()),
                   new String(record.key(), StandardCharsets.UTF_8),
                   value,
-                  record.timestamp(),
+                  Optional.of(record.timestamp()),
                   null
               ),
               null)

ksql-functional-tests/src/main/java/io/confluent/ksql/test/TestFrameworkException.java

@@ -23,4 +23,8 @@ public class TestFrameworkException extends RuntimeException {
   public TestFrameworkException(final String msg) {
     super(msg);
   }
+
+  public TestFrameworkException(final String msg, final Throwable cause) {
+    super(msg, cause);
+  }
 }