KAFKA-7461: Add tests for logical types

Added testing of logical types for Kafka Connect in support of KIP-145 features.
Added tests for Boolean, Time, Date and Timestamp, including the valid conversions.

The area of ISO8601 strings is a bit of a mess because the tokenizer is not compatible with
that format, and a subsequent JIRA will be needed to fix that.

A few small fixes as well as creating test cases, but they're clearly just corrections such as
using 0 to mean January (java.util.Calendar uses zero-based month numbers).

Author: Andrew Schofield <andrew_schofield@uk.ibm.com>

Reviewers: Mickael Maison <mimaison@users.noreply.github.com>, Ewen Cheslack-Postava <ewen@confluent.io>

Closes #6077 from AndrewJSchofield/KAFKA-7461-ConverterValuesLogicalTypesTest

connect/api/src/main/java/org/apache/kafka/connect/data/Values.java

@@ -67,7 +67,7 @@ public class Values {
     private static final Schema MAP_SELECTOR_SCHEMA = SchemaBuilder.map(Schema.STRING_SCHEMA, Schema.STRING_SCHEMA).build();
     private static final Schema STRUCT_SELECTOR_SCHEMA = SchemaBuilder.struct().build();
     private static final String TRUE_LITERAL = Boolean.TRUE.toString();
-    private static final String FALSE_LITERAL = Boolean.TRUE.toString();
+    private static final String FALSE_LITERAL = Boolean.FALSE.toString();
     private static final long MILLIS_PER_DAY = 24 * 60 * 60 * 1000;
     private static final String NULL_VALUE = "null";
     private static final String ISO_8601_DATE_FORMAT_PATTERN = "yyyy-MM-dd";
@@ -488,7 +488,7 @@ public class Values {
                                 Calendar calendar = Calendar.getInstance(UTC);
                                 calendar.setTime((java.util.Date) value);
                                 calendar.set(Calendar.YEAR, 1970);
-                                calendar.set(Calendar.MONTH, 1);
+                                calendar.set(Calendar.MONTH, 0); // Months are zero-based
                                 calendar.set(Calendar.DAY_OF_MONTH, 1);
                                 return Time.toLogical(toSchema, (int) calendar.getTimeInMillis());
                             }
@@ -872,7 +872,7 @@ public class Values {
                 }
             } else if (tokenLength == ISO_8601_TIMESTAMP_LENGTH) {
                 try {
-                    return new SchemaAndValue(Time.SCHEMA, new SimpleDateFormat(ISO_8601_TIMESTAMP_FORMAT_PATTERN).parse(token));
+                    return new SchemaAndValue(Timestamp.SCHEMA, new SimpleDateFormat(ISO_8601_TIMESTAMP_FORMAT_PATTERN).parse(token));
                 } catch (ParseException e) {
                     // not a valid date
                 }

connect/api/src/test/java/org/apache/kafka/connect/data/ValuesTest.java

@@ -35,6 +35,8 @@ import static org.junit.Assert.fail;
 
 public class ValuesTest {
 
+    private static final long MILLIS_PER_DAY = 24 * 60 * 60 * 1000;
+
     private static final Map<String, String> STRING_MAP = new LinkedHashMap<>();
     private static final Schema STRING_MAP_SCHEMA = SchemaBuilder.map(Schema.STRING_SCHEMA, Schema.STRING_SCHEMA).schema();
 
@@ -78,6 +80,24 @@ public class ValuesTest {
         assertRoundTrip(Schema.OPTIONAL_STRING_SCHEMA, Schema.STRING_SCHEMA, null);
     }
 
+    @Test
+    public void shouldConvertBooleanValues() {
+        assertRoundTrip(Schema.BOOLEAN_SCHEMA, Schema.BOOLEAN_SCHEMA, Boolean.FALSE);
+        SchemaAndValue resultFalse = roundTrip(Schema.BOOLEAN_SCHEMA, "false");
+        assertEquals(Schema.BOOLEAN_SCHEMA, resultFalse.schema());
+        assertEquals(Boolean.FALSE, resultFalse.value());
+
+        assertRoundTrip(Schema.BOOLEAN_SCHEMA, Schema.BOOLEAN_SCHEMA, Boolean.TRUE);
+        SchemaAndValue resultTrue = roundTrip(Schema.BOOLEAN_SCHEMA, "true");
+        assertEquals(Schema.BOOLEAN_SCHEMA, resultTrue.schema());
+        assertEquals(Boolean.TRUE, resultTrue.value());
+    }
+
+    @Test(expected = DataException.class)
+    public void shouldFailToParseInvalidBooleanValueString() {
+        Values.convertToBoolean(Schema.STRING_SCHEMA, "\"green\"");
+    }
+
     @Test
     public void shouldConvertSimpleString() {
         assertRoundTrip(Schema.STRING_SCHEMA,  "simple");
@@ -327,6 +347,85 @@ public class ValuesTest {
         assertParsed("a { b c } d", "a ", "{", " b c ", "}", " d");
     }
 
+    @Test
+    public void shouldConvertTimeValues() {
+        java.util.Date current = new java.util.Date();
+        long currentMillis = current.getTime() % MILLIS_PER_DAY;
+
+        // java.util.Date - just copy
+        java.util.Date t1 = Values.convertToTime(Time.SCHEMA, current);
+        assertEquals(current, t1);
+
+        // java.util.Date as a Timestamp - discard the date and keep just day's milliseconds
+        t1 = Values.convertToTime(Timestamp.SCHEMA, current);
+        assertEquals(new java.util.Date(currentMillis), t1);
+
+        // ISO8601 strings - currently broken because tokenization breaks at colon
+
+        // Millis as string
+        java.util.Date t3 = Values.convertToTime(Time.SCHEMA, Long.toString(currentMillis));
+        assertEquals(currentMillis, t3.getTime());
+
+        // Millis as long
+        java.util.Date t4 = Values.convertToTime(Time.SCHEMA, currentMillis);
+        assertEquals(currentMillis, t4.getTime());
+    }
+
+    @Test
+    public void shouldConvertDateValues() {
+        java.util.Date current = new java.util.Date();
+        long currentMillis = current.getTime() % MILLIS_PER_DAY;
+        long days = current.getTime() / MILLIS_PER_DAY;
+
+        // java.util.Date - just copy
+        java.util.Date d1 = Values.convertToDate(Date.SCHEMA, current);
+        assertEquals(current, d1);
+
+        // java.util.Date as a Timestamp - discard the day's milliseconds and keep the date
+        java.util.Date currentDate = new java.util.Date(current.getTime() - currentMillis);
+        d1 = Values.convertToDate(Timestamp.SCHEMA, currentDate);
+        assertEquals(currentDate, d1);
+
+        // ISO8601 strings - currently broken because tokenization breaks at colon
+
+        // Days as string
+        java.util.Date d3 = Values.convertToDate(Date.SCHEMA, Long.toString(days));
+        assertEquals(currentDate, d3);
+
+        // Days as long
+        java.util.Date d4 = Values.convertToDate(Date.SCHEMA, days);
+        assertEquals(currentDate, d4);
+    }
+
+    @Test
+    public void shouldConvertTimestampValues() {
+        java.util.Date current = new java.util.Date();
+        long currentMillis = current.getTime() % MILLIS_PER_DAY;
+
+        // java.util.Date - just copy
+        java.util.Date ts1 = Values.convertToTimestamp(Timestamp.SCHEMA, current);
+        assertEquals(current, ts1);
+
+        // java.util.Date as a Timestamp - discard the day's milliseconds and keep the date
+        java.util.Date currentDate = new java.util.Date(current.getTime() - currentMillis);
+        ts1 = Values.convertToTimestamp(Date.SCHEMA, currentDate);
+        assertEquals(currentDate, ts1);
+
+        // java.util.Date as a Time - discard the date and keep the day's milliseconds
+        ts1 = Values.convertToTimestamp(Time.SCHEMA, currentMillis);
+        assertEquals(new java.util.Date(currentMillis), ts1);
+
+        // ISO8601 strings - currently broken because tokenization breaks at colon
+
+        // Millis as string
+        java.util.Date ts3 = Values.convertToTimestamp(Timestamp.SCHEMA, Long.toString(current.getTime()));
+        assertEquals(current, ts3);
+
+        // Millis as long
+        java.util.Date ts4 = Values.convertToTimestamp(Timestamp.SCHEMA, current.getTime());
+        assertEquals(current, ts4);
+    }
+
     @Test
     public void canConsume() {
     }
@@ -384,7 +483,6 @@ public class ValuesTest {
         return roundTrip(desiredSchema, new SchemaAndValue(Schema.STRING_SCHEMA, currentValue));
     }
 
-
     protected SchemaAndValue roundTrip(Schema desiredSchema, SchemaAndValue input) {
         String serialized = Values.convertToString(input.schema(), input.value());
         if (input != null && input.value() != null) {
@@ -458,5 +556,4 @@ public class ValuesTest {
             assertEquals(result, result2);
         }
     }
-
 }