more tests

Author: jerrinot

connector/src/test/java/io/questdb/kafka/QuestDBSinkConnectorEmbeddedTest.java

@@ -114,6 +114,35 @@ public void testSmoke() {
                 "select firstname,lastname,age from " + topicName);
     }
 
+    @Test
+    public void testUpfrontTable() {
+        connect.kafka().createTopic(topicName, 1);
+        Map<String, String> props = baseConnectorProps(topicName);
+        connect.configureConnector(CONNECTOR_NAME, props);
+        assertConnectorTaskRunningEventually();
+        Schema schema = SchemaBuilder.struct().name("com.example.Person")
+                .field("firstname", Schema.STRING_SCHEMA)
+                .field("lastname", Schema.STRING_SCHEMA)
+                .field("age", Schema.INT8_SCHEMA)
+                .build();
+
+        Struct struct = new Struct(schema)
+                .put("firstname", "John")
+                .put("lastname", "Doe")
+                .put("age", (byte) 42);
+
+        QuestDBUtils.assertSql(questDBContainer,
+                "{\"ddl\":\"OK\"}\n",
+                "create table " + topicName + " (firstname string, lastname string, age int)",
+                QuestDBUtils.Endpoint.EXEC);
+
+        connect.kafka().produce(topicName, "key", new String(converter.fromConnectData(topicName, schema, struct)));
+
+        QuestDBUtils.assertSqlEventually(questDBContainer, "\"firstname\",\"lastname\",\"age\",\"key\"\r\n"
+                        + "\"John\",\"Doe\",42,\"key\"\r\n",
+                "select * from " + topicName);
+    }
+
     @Test
     public void testDesignatedTimestamp_noSchema_unixEpochMillis() {
         connect.kafka().createTopic(topicName, 1);

connector/src/test/java/io/questdb/kafka/QuestDBUtils.java

@@ -1,24 +1,43 @@
 package io.questdb.kafka;
 
+import okhttp3.ConnectionSpec;
 import okhttp3.OkHttpClient;
 import okhttp3.Request;
 import okhttp3.Response;
 import org.testcontainers.containers.GenericContainer;
 
 import java.io.IOException;
 import java.net.URLEncoder;
+import java.util.Collections;
 import java.util.concurrent.TimeUnit;
 
 import static org.awaitility.Awaitility.await;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.fail;
 
 public final class QuestDBUtils {
+    public enum Endpoint {
+        EXPORT("exp"),
+        EXEC("exec");
+
+        private String endpoint;
+
+        Endpoint(String endpoint) {
+            this.endpoint = endpoint;
+        }
+
+
+        String getEndpoint() {
+            return endpoint;
+        }
+    }
+
+
     public static final int QUESTDB_ILP_PORT = 9009;
     public static final int QUESTDB_HTTP_PORT = 9000;
 
     private static final int QUERY_WAITING_TIME_SECONDS = 30;
-    private static final OkHttpClient CLIENT = new OkHttpClient();
+    private static final OkHttpClient CLIENT = new OkHttpClient(new OkHttpClient.Builder().connectionSpecs(Collections.singletonList(ConnectionSpec.CLEARTEXT)));
 
     private QuestDBUtils() {
 
@@ -28,8 +47,12 @@ public static void assertSqlEventually(GenericContainer<?> questdbContainer, Str
         await().atMost(QUERY_WAITING_TIME_SECONDS, TimeUnit.SECONDS).untilAsserted(() -> assertSql(questdbContainer, expectedResult, query));
     }
 
-    private static void assertSql(GenericContainer<?> questdbContainer, String expectedResult, String query) throws IOException {
-        try (Response response = executeQuery(questdbContainer, query)) {
+    public static void assertSql(GenericContainer<?> questdbContainer, String expectedResult, String query) {
+        assertSql(questdbContainer, expectedResult, query, Endpoint.EXPORT);
+    }
+
+    public static void assertSql(GenericContainer<?> questdbContainer, String expectedResult, String query, Endpoint endpoint) {
+        try (Response response = executeQuery(questdbContainer, query, endpoint)) {
             if (response.code() != 200) {
                 fail("Query failed, returned code " + response.code());
             }
@@ -39,14 +62,16 @@ private static void assertSql(GenericContainer<?> questdbContainer, String expec
                     assertEquals(expectedResult, bodyString);
                 }
             }
+        } catch (IOException e) {
+            fail("Query failed", e);
         }
     }
 
-    private static Response executeQuery(GenericContainer<?> questContainer, String query) throws IOException {
+    private static Response executeQuery(GenericContainer<?> questContainer, String query, Endpoint endpoint) throws IOException {
         String encodedQuery = URLEncoder.encode(query, "UTF-8");
         String baseUrl = "http://" + questContainer.getHost() + ":" + questContainer.getMappedPort(QUESTDB_HTTP_PORT);
         Request request = new Request.Builder()
-                .url(baseUrl + "/exp?query=" + encodedQuery)
+                .url(baseUrl + "/" + endpoint.endpoint + "?query=" + encodedQuery)
                 .build();
         return CLIENT.newCall(request).execute();
     }

integration-tests/debezium/src/test/java/kafka/DebeziumIT.java

@@ -322,6 +322,36 @@ public void testDate() throws SQLException {
         }
     }
 
+    @Test
+    public void testDelete() throws SQLException {
+        String questTableName = "test_delete";
+        try (Connection connection = getConnection(postgresContainer);
+             Statement statement = connection.createStatement()) {
+            startDebeziumConnector();
+
+            statement.execute("create schema " + PG_SCHEMA_NAME);
+            statement.execute("create table " + PG_SCHEMA_NAME + "." + PG_TABLE_NAME + " (id int8 not null, title varchar(255), created_at timestamp, primary key (id))");
+            statement.execute("insert into "  + PG_SCHEMA_NAME + "." + PG_TABLE_NAME + " values (1, 'Learn CDC', '2021-01-02')");
+
+            ConnectorConfiguration questSink = newQuestSinkBaseConfig(questTableName);
+            questSink.with(QuestDBSinkConnectorConfig.DESIGNATED_TIMESTAMP_COLUMN_NAME_CONFIG, "created_at");
+            debeziumContainer.registerConnector(QUESTDB_CONNECTOR_NAME, questSink);
+
+            QuestDBUtils.assertSqlEventually(questDBContainer, "\"id\",\"title\",\"timestamp\"\r\n"
+                            + "1,\"Learn CDC\",\"2021-01-02T00:00:00.000000Z\"\r\n",
+                    "select * from " + questTableName);
+
+            // delete should be ignored by QuestDB
+            statement.execute("delete from "  + PG_SCHEMA_NAME + "." + PG_TABLE_NAME + " where id = 1");
+            statement.execute("insert into "  + PG_SCHEMA_NAME + "." + PG_TABLE_NAME + " values (2, 'Learn Debezium', '2021-01-03')");
+
+            QuestDBUtils.assertSqlEventually(questDBContainer, "\"id\",\"title\",\"timestamp\"\r\n"
+                            + "1,\"Learn CDC\",\"2021-01-02T00:00:00.000000Z\"\r\n"
+                            + "2,\"Learn Debezium\",\"2021-01-03T00:00:00.000000Z\"\r\n",
+                    "select * from " + questTableName);
+        }
+    }
+
     private static void startDebeziumConnector() {
         ConnectorConfiguration connector = ConnectorConfiguration
                 .forJdbcContainer(postgresContainer)