test: add DATE/TIME to connect integration test

ksqldb-rest-app/src/test/java/io/confluent/ksql/rest/server/ConnectIntegrationTest.java

@@ -255,7 +255,7 @@ public void shouldReturnError() {
   }
 
   @Test
-  public void shouldReadTimestampsFromConnect() {
+  public void shouldReadTimeTypesFromConnect() {
     // Given:
     create("mock-source", ImmutableMap.<String, String> builder()
         .put("connector.class", "org.apache.kafka.connect.tools.VerifiableSourceConnector")
@@ -266,18 +266,22 @@ public void shouldReadTimestampsFromConnect() {
         .put("topic.creation.default.partitions", "1")
         .build());
 
-    final long start = System.nanoTime();
-    RestResponse<KsqlEntityList> response;
-    do {
-      response = ksqlRestClient.makeKsqlRequest("CREATE STREAM FOO (PAYLOAD TIMESTAMP) WITH (KAFKA_TOPIC='foo', VALUE_FORMAT='DELIMITED');");
-    } while(!response.isSuccessful() && System.nanoTime() - start < TIMEOUT_NS);
+    makeKsqlRequest("CREATE STREAM TIMESTAMP_STREAM (PAYLOAD TIMESTAMP) WITH (KAFKA_TOPIC='foo', VALUE_FORMAT='DELIMITED');");
+    makeKsqlRequest("CREATE STREAM TIME_STREAM (PAYLOAD TIME) WITH (KAFKA_TOPIC='foo', VALUE_FORMAT='DELIMITED');");
+    makeKsqlRequest("CREATE STREAM DATE_STREAM (PAYLOAD DATE) WITH (KAFKA_TOPIC='foo', VALUE_FORMAT='DELIMITED');");
 
     // When:
-    final RestResponse<List<StreamedRow>> queryFoo = ksqlRestClient.makeQueryRequest("SELECT * FROM FOO EMIT CHANGES LIMIT 1;", 1L);
+    final RestResponse<List<StreamedRow>> queryTimestamp = ksqlRestClient.makeQueryRequest("SELECT * FROM TIMESTAMP_STREAM EMIT CHANGES LIMIT 1;", 1L);
+    final RestResponse<List<StreamedRow>> queryTime = ksqlRestClient.makeQueryRequest("SELECT * FROM TIME_STREAM EMIT CHANGES LIMIT 1;", 1L);
+    final RestResponse<List<StreamedRow>> queryDate = ksqlRestClient.makeQueryRequest("SELECT * FROM DATE_STREAM EMIT CHANGES LIMIT 1;", 1L);
 
     // Then:
-    assertThat("successfully queried FOO", queryFoo.isSuccessful());
-    assertThat(queryFoo.getResponse().get(1).getRow().get().getColumns().get(0), is("1970-01-01T00:00:00.000"));
+    assertThat("successfully queried TIMESTAMP_STREAM", queryTimestamp.isSuccessful());
+    assertThat("successfully queried TIME_STREAM", queryTime.isSuccessful());
+    assertThat("successfully queried DATE_STREAM", queryDate.isSuccessful());
+    assertThat(queryTimestamp.getResponse().get(1).getRow().get().getColumns().get(0), is("1970-01-01T00:00:00.000"));
+    assertThat(queryTime.getResponse().get(1).getRow().get().getColumns().get(0), is("00:00"));
+    assertThat(queryDate.getResponse().get(1).getRow().get().getColumns().get(0), is("1970-01-01"));
   }
 
   @Test
@@ -330,4 +334,11 @@ private void create(final String name, final Map<String, String> properties, Con
     LOG.info("Got response from Connect: {}", response);
   }
 
+  private void makeKsqlRequest(final String request) {
+    final long start = System.nanoTime();
+    RestResponse<KsqlEntityList> response;
+    do {
+      response = ksqlRestClient.makeKsqlRequest(request);
+    } while(!response.isSuccessful() && System.nanoTime() - start < TIMEOUT_NS);
+  }
 }

ksqldb-rest-model/pom.xml

@@ -35,6 +35,12 @@
             <version>${io.confluent.ksql.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>io.confluent.ksql</groupId>
+            <artifactId>ksqldb-engine-common</artifactId>
+            <version>${io.confluent.ksql.version}</version>
+        </dependency>
+
         <dependency>
             <groupId>javax.ws.rs</groupId>
             <artifactId>javax.ws.rs-api</artifactId>

ksqldb-rest-model/src/main/java/io/confluent/ksql/rest/ApiJsonMapper.java

@@ -17,20 +17,29 @@
 
 import com.fasterxml.jackson.core.JsonGenerator;
 import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.JsonSerializer;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.module.SimpleModule;
 import com.fasterxml.jackson.databind.node.JsonNodeFactory;
 import com.fasterxml.jackson.datatype.guava.GuavaModule;
 import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
 import io.confluent.ksql.json.KsqlTypesSerializationModule;
 import io.confluent.ksql.json.StructSerializationModule;
+import io.confluent.ksql.schema.ksql.SqlTimeTypes;
 import io.confluent.ksql.util.KsqlConstants;
+import java.io.IOException;
+import java.sql.Date;
+import java.sql.Time;
 import java.text.SimpleDateFormat;
 import java.util.TimeZone;
 
 /**
  * Mapper used by the Rest Api.
  */
+// CHECKSTYLE_RULES.OFF: ClassDataAbstractionCoupling
 public enum ApiJsonMapper {
+  // CHECKSTYLE_RULES.ON: ClassDataAbstractionCoupling
 
   INSTANCE;
 
@@ -46,9 +55,29 @@ public enum ApiJsonMapper {
       .enable(JsonGenerator.Feature.WRITE_BIGDECIMAL_AS_PLAIN)
       .setDateFormat(new SimpleDateFormat(KsqlConstants.DATE_TIME_PATTERN))
       .setTimeZone(TimeZone.getTimeZone("Z"))
-      .setNodeFactory(JsonNodeFactory.withExactBigDecimals(true));
+      .setNodeFactory(JsonNodeFactory.withExactBigDecimals(true))
+      .registerModule(new SimpleModule()
+          .addSerializer(Time.class, new TimeSerializer())
+          .addSerializer(Date.class, new DateSerializer())
+      );
 
   public ObjectMapper get() {
     return mapper;
   }
+
+  public static class TimeSerializer extends JsonSerializer<Time> {
+    @Override
+    public void serialize(final Time time, final JsonGenerator jsonGenerator,
+        final SerializerProvider serializerProvider) throws IOException {
+      jsonGenerator.writeString(SqlTimeTypes.formatTime(time));
+    }
+  }
+
+  public static class DateSerializer extends JsonSerializer<Date> {
+    @Override
+    public void serialize(final Date date, final JsonGenerator jsonGenerator,
+        final SerializerProvider serializerProvider) throws IOException {
+      jsonGenerator.writeString(SqlTimeTypes.formatDate(date));
+    }
+  }
 }

ksqldb-rest-model/src/test/java/io/confluent/ksql/rest/ApiJsonMapperTest.java

@@ -25,6 +25,8 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.DecimalNode;
 import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Time;
 import java.sql.Timestamp;
 import org.junit.Test;
 
@@ -79,4 +81,22 @@ public void shouldFormatTimestampsToISO8601() throws Exception {
     // Then:
     assertThat(result, is("\"1999-11-30T11:00:00.000\""));
   }
+
+  @Test
+  public void shouldFormatTime() throws Exception {
+    // When:
+    final String result = OBJECT_MAPPER.writeValueAsString(new Time(10000));
+
+    // Then:
+    assertThat(result, is("\"00:00:10\""));
+  }
+
+  @Test
+  public void shouldFormatDate() throws Exception {
+    // When:
+    final String result = OBJECT_MAPPER.writeValueAsString(new Date(864000000));
+
+    // Then:
+    assertThat(result, is("\"1970-01-11\""));
+  }
 }