diff --git a/connect/api/src/main/java/org/apache/kafka/connect/data/Values.java b/connect/api/src/main/java/org/apache/kafka/connect/data/Values.java index 93c320a236760..d99fbcabf86df 100644 --- a/connect/api/src/main/java/org/apache/kafka/connect/data/Values.java +++ b/connect/api/src/main/java/org/apache/kafka/connect/data/Values.java @@ -30,13 +30,17 @@ import java.text.SimpleDateFormat; import java.text.StringCharacterIterator; import java.util.ArrayList; +import java.util.Arrays; import java.util.Base64; import java.util.Calendar; +import java.util.Collections; +import java.util.HashSet; import java.util.Iterator; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.NoSuchElementException; +import java.util.Set; import java.util.TimeZone; import java.util.regex.Pattern; @@ -70,9 +74,18 @@ public class Values { 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"; - private static final String ISO_8601_TIME_FORMAT_PATTERN = "HH:mm:ss.SSS'Z'"; - private static final String ISO_8601_TIMESTAMP_FORMAT_PATTERN = ISO_8601_DATE_FORMAT_PATTERN + "'T'" + ISO_8601_TIME_FORMAT_PATTERN; + static final String ISO_8601_DATE_FORMAT_PATTERN = "yyyy-MM-dd"; + static final String ISO_8601_TIME_FORMAT_PATTERN = "HH:mm:ss.SSS'Z'"; + static final String ISO_8601_TIMESTAMP_FORMAT_PATTERN = ISO_8601_DATE_FORMAT_PATTERN + "'T'" + ISO_8601_TIME_FORMAT_PATTERN; + private static final Set TEMPORAL_LOGICAL_TYPE_NAMES = + Collections.unmodifiableSet( + new HashSet<>( + Arrays.asList(Time.LOGICAL_NAME, + Timestamp.LOGICAL_NAME, + Date.LOGICAL_NAME + ) + ) + ); private static final String QUOTE_DELIMITER = "\""; private static final String COMMA_DELIMITER = ","; @@ -467,6 +480,9 @@ protected static Object convertTo(Schema toSchema, Schema fromSchema, Object val int days = (int) (millis / MILLIS_PER_DAY); // truncates return Date.toLogical(toSchema, days); } + } else { + // There is no fromSchema, so no conversion is needed + return value; } } long numeric = asLong(value, fromSchema, null); @@ -492,6 +508,9 @@ protected static Object convertTo(Schema toSchema, Schema fromSchema, Object val calendar.set(Calendar.DAY_OF_MONTH, 1); return Time.toLogical(toSchema, (int) calendar.getTimeInMillis()); } + } else { + // There is no fromSchema, so no conversion is needed + return value; } } long numeric = asLong(value, fromSchema, null); @@ -523,6 +542,9 @@ protected static Object convertTo(Schema toSchema, Schema fromSchema, Object val if (Timestamp.LOGICAL_NAME.equals(fromSchemaName)) { return value; } + } else { + // There is no fromSchema, so no conversion is needed + return value; } } long numeric = asLong(value, fromSchema, null); @@ -755,7 +777,14 @@ protected static SchemaAndValue parse(Parser parser, boolean embedded) throws No } sb.append(parser.next()); } - return new SchemaAndValue(Schema.STRING_SCHEMA, sb.toString()); + String content = sb.toString(); + // We can parse string literals as temporal logical types, but all others + // are treated as strings + SchemaAndValue parsed = parseString(content); + if (parsed != null && TEMPORAL_LOGICAL_TYPE_NAMES.contains(parsed.schema().name())) { + return parsed; + } + return new SchemaAndValue(Schema.STRING_SCHEMA, content); } } @@ -838,7 +867,9 @@ protected static SchemaAndValue parse(Parser parser, boolean embedded) throws No } if (!parser.canConsume(ENTRY_DELIMITER)) { - throw new DataException("Map entry is missing '=': " + parser.original()); + throw new DataException("Map entry is missing '" + ENTRY_DELIMITER + + "' at " + parser.position() + + " in " + parser.original()); } SchemaAndValue value = parse(parser, true); Object entryValue = value != null ? value.value() : null; @@ -855,7 +886,7 @@ protected static SchemaAndValue parse(Parser parser, boolean embedded) throws No throw new DataException("Map is missing terminating '}': " + parser.original()); } } catch (DataException e) { - LOG.debug("Unable to parse the value as a map or an array; reverting to string", e); + LOG.trace("Unable to parse the value as a map or an array; reverting to string", e); parser.rewindTo(startPosition); } @@ -867,6 +898,27 @@ protected static SchemaAndValue parse(Parser parser, boolean embedded) throws No char firstChar = token.charAt(0); boolean firstCharIsDigit = Character.isDigit(firstChar); + + // Temporal types are more restrictive, so try them first + if (firstCharIsDigit) { + // The time and timestamp literals may be split into 5 tokens since an unescaped colon + // is a delimiter. Check these first since the first of these tokens is a simple numeric + int position = parser.mark(); + String remainder = parser.next(4); + if (remainder != null) { + String timeOrTimestampStr = token + remainder; + SchemaAndValue temporal = parseAsTemporal(timeOrTimestampStr); + if (temporal != null) { + return temporal; + } + } + // No match was found using the 5 tokens, so rewind and see if the current token has a date, time, or timestamp + parser.rewindTo(position); + SchemaAndValue temporal = parseAsTemporal(token); + if (temporal != null) { + return temporal; + } + } if (firstCharIsDigit || firstChar == '+' || firstChar == '-') { try { // Try to parse as a number ... @@ -901,37 +953,42 @@ protected static SchemaAndValue parse(Parser parser, boolean embedded) throws No // can't parse as a number } } - if (firstCharIsDigit) { - // Check for a date, time, or timestamp ... - int tokenLength = token.length(); - if (tokenLength == ISO_8601_DATE_LENGTH) { - try { - return new SchemaAndValue(Date.SCHEMA, new SimpleDateFormat(ISO_8601_DATE_FORMAT_PATTERN).parse(token)); - } catch (ParseException e) { - // not a valid date - } - } else if (tokenLength == ISO_8601_TIME_LENGTH) { - try { - return new SchemaAndValue(Time.SCHEMA, new SimpleDateFormat(ISO_8601_TIME_FORMAT_PATTERN).parse(token)); - } catch (ParseException e) { - // not a valid date - } - } else if (tokenLength == ISO_8601_TIMESTAMP_LENGTH) { - try { - return new SchemaAndValue(Timestamp.SCHEMA, new SimpleDateFormat(ISO_8601_TIMESTAMP_FORMAT_PATTERN).parse(token)); - } catch (ParseException e) { - // not a valid date - } - } - } if (embedded) { throw new DataException("Failed to parse embedded value"); } - // At this point, the only thing this can be is a string. Embedded strings were processed above, - // so this is not embedded and we can use the original string... + // At this point, the only thing this non-embedded value can be is a string. return new SchemaAndValue(Schema.STRING_SCHEMA, parser.original()); } + private static SchemaAndValue parseAsTemporal(String token) { + if (token == null) { + return null; + } + // If the colons were escaped, we'll see the escape chars and need to remove them + token = token.replace("\\:", ":"); + int tokenLength = token.length(); + if (tokenLength == ISO_8601_TIME_LENGTH) { + try { + return new SchemaAndValue(Time.SCHEMA, new SimpleDateFormat(ISO_8601_TIME_FORMAT_PATTERN).parse(token)); + } catch (ParseException e) { + // not a valid date + } + } else if (tokenLength == ISO_8601_TIMESTAMP_LENGTH) { + try { + return new SchemaAndValue(Timestamp.SCHEMA, new SimpleDateFormat(ISO_8601_TIMESTAMP_FORMAT_PATTERN).parse(token)); + } catch (ParseException e) { + // not a valid date + } + } else if (tokenLength == ISO_8601_DATE_LENGTH) { + try { + return new SchemaAndValue(Date.SCHEMA, new SimpleDateFormat(ISO_8601_DATE_FORMAT_PATTERN).parse(token)); + } catch (ParseException e) { + // not a valid date + } + } + return null; + } + protected static Schema commonSchemaFor(Schema previous, SchemaAndValue latest) { if (latest == null) { return previous; @@ -1088,6 +1145,7 @@ public int mark() { public void rewindTo(int position) { iter.setIndex(position); nextToken = null; + previousToken = null; } public String original() { @@ -1112,6 +1170,19 @@ public String next() { return previousToken; } + public String next(int n) { + int current = mark(); + int start = mark(); + for (int i = 0; i != n; ++i) { + if (!hasNext()) { + rewindTo(start); + return null; + } + next(); + } + return original.substring(current, position()); + } + private String consumeNextToken() throws NoSuchElementException { boolean escaped = false; int start = iter.getIndex(); diff --git a/connect/api/src/test/java/org/apache/kafka/connect/data/ValuesTest.java b/connect/api/src/test/java/org/apache/kafka/connect/data/ValuesTest.java index a5909f3ece9bf..c437e46c25956 100644 --- a/connect/api/src/test/java/org/apache/kafka/connect/data/ValuesTest.java +++ b/connect/api/src/test/java/org/apache/kafka/connect/data/ValuesTest.java @@ -21,6 +21,7 @@ import org.apache.kafka.connect.errors.DataException; import org.junit.Test; +import java.text.SimpleDateFormat; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -383,6 +384,146 @@ public void shouldParseStringListWithExtraDelimitersAndReturnString() { assertEquals(str, result.value()); } + @Test + public void shouldParseTimestampStringAsTimestamp() throws Exception { + String str = "2019-08-23T14:34:54.346Z"; + SchemaAndValue result = Values.parseString(str); + assertEquals(Type.INT64, result.schema().type()); + assertEquals(Timestamp.LOGICAL_NAME, result.schema().name()); + java.util.Date expected = new SimpleDateFormat(Values.ISO_8601_TIMESTAMP_FORMAT_PATTERN).parse(str); + assertEquals(expected, result.value()); + } + + @Test + public void shouldParseDateStringAsDate() throws Exception { + String str = "2019-08-23"; + SchemaAndValue result = Values.parseString(str); + assertEquals(Type.INT32, result.schema().type()); + assertEquals(Date.LOGICAL_NAME, result.schema().name()); + java.util.Date expected = new SimpleDateFormat(Values.ISO_8601_DATE_FORMAT_PATTERN).parse(str); + assertEquals(expected, result.value()); + } + + @Test + public void shouldParseTimeStringAsDate() throws Exception { + String str = "14:34:54.346Z"; + SchemaAndValue result = Values.parseString(str); + assertEquals(Type.INT32, result.schema().type()); + assertEquals(Time.LOGICAL_NAME, result.schema().name()); + java.util.Date expected = new SimpleDateFormat(Values.ISO_8601_TIME_FORMAT_PATTERN).parse(str); + assertEquals(expected, result.value()); + } + + @Test + public void shouldParseTimestampStringWithEscapedColonsAsTimestamp() throws Exception { + String str = "2019-08-23T14\\:34\\:54.346Z"; + SchemaAndValue result = Values.parseString(str); + assertEquals(Type.INT64, result.schema().type()); + assertEquals(Timestamp.LOGICAL_NAME, result.schema().name()); + String expectedStr = "2019-08-23T14:34:54.346Z"; + java.util.Date expected = new SimpleDateFormat(Values.ISO_8601_TIMESTAMP_FORMAT_PATTERN).parse(expectedStr); + assertEquals(expected, result.value()); + } + + @Test + public void shouldParseTimeStringWithEscapedColonsAsDate() throws Exception { + String str = "14\\:34\\:54.346Z"; + SchemaAndValue result = Values.parseString(str); + assertEquals(Type.INT32, result.schema().type()); + assertEquals(Time.LOGICAL_NAME, result.schema().name()); + String expectedStr = "14:34:54.346Z"; + java.util.Date expected = new SimpleDateFormat(Values.ISO_8601_TIME_FORMAT_PATTERN).parse(expectedStr); + assertEquals(expected, result.value()); + } + + @Test + public void shouldParseDateStringAsDateInArray() throws Exception { + String dateStr = "2019-08-23"; + String arrayStr = "[" + dateStr + "]"; + SchemaAndValue result = Values.parseString(arrayStr); + assertEquals(Type.ARRAY, result.schema().type()); + Schema elementSchema = result.schema().valueSchema(); + assertEquals(Type.INT32, elementSchema.type()); + assertEquals(Date.LOGICAL_NAME, elementSchema.name()); + java.util.Date expected = new SimpleDateFormat(Values.ISO_8601_DATE_FORMAT_PATTERN).parse(dateStr); + assertEquals(Collections.singletonList(expected), result.value()); + } + + @Test + public void shouldParseTimeStringAsTimeInArray() throws Exception { + String timeStr = "14:34:54.346Z"; + String arrayStr = "[" + timeStr + "]"; + SchemaAndValue result = Values.parseString(arrayStr); + assertEquals(Type.ARRAY, result.schema().type()); + Schema elementSchema = result.schema().valueSchema(); + assertEquals(Type.INT32, elementSchema.type()); + assertEquals(Time.LOGICAL_NAME, elementSchema.name()); + java.util.Date expected = new SimpleDateFormat(Values.ISO_8601_TIME_FORMAT_PATTERN).parse(timeStr); + assertEquals(Collections.singletonList(expected), result.value()); + } + + @Test + public void shouldParseTimestampStringAsTimestampInArray() throws Exception { + String tsStr = "2019-08-23T14:34:54.346Z"; + String arrayStr = "[" + tsStr + "]"; + SchemaAndValue result = Values.parseString(arrayStr); + assertEquals(Type.ARRAY, result.schema().type()); + Schema elementSchema = result.schema().valueSchema(); + assertEquals(Type.INT64, elementSchema.type()); + assertEquals(Timestamp.LOGICAL_NAME, elementSchema.name()); + java.util.Date expected = new SimpleDateFormat(Values.ISO_8601_TIMESTAMP_FORMAT_PATTERN).parse(tsStr); + assertEquals(Collections.singletonList(expected), result.value()); + } + + @Test + public void shouldParseMultipleTimestampStringAsTimestampInArray() throws Exception { + String tsStr1 = "2019-08-23T14:34:54.346Z"; + String tsStr2 = "2019-01-23T15:12:34.567Z"; + String tsStr3 = "2019-04-23T19:12:34.567Z"; + String arrayStr = "[" + tsStr1 + "," + tsStr2 + ", " + tsStr3 + "]"; + SchemaAndValue result = Values.parseString(arrayStr); + assertEquals(Type.ARRAY, result.schema().type()); + Schema elementSchema = result.schema().valueSchema(); + assertEquals(Type.INT64, elementSchema.type()); + assertEquals(Timestamp.LOGICAL_NAME, elementSchema.name()); + java.util.Date expected1 = new SimpleDateFormat(Values.ISO_8601_TIMESTAMP_FORMAT_PATTERN).parse(tsStr1); + java.util.Date expected2 = new SimpleDateFormat(Values.ISO_8601_TIMESTAMP_FORMAT_PATTERN).parse(tsStr2); + java.util.Date expected3 = new SimpleDateFormat(Values.ISO_8601_TIMESTAMP_FORMAT_PATTERN).parse(tsStr3); + assertEquals(Arrays.asList(expected1, expected2, expected3), result.value()); + } + + @Test + public void shouldParseQuotedTimeStringAsTimeInMap() throws Exception { + String keyStr = "k1"; + String timeStr = "14:34:54.346Z"; + String mapStr = "{\"" + keyStr + "\":\"" + timeStr + "\"}"; + SchemaAndValue result = Values.parseString(mapStr); + assertEquals(Type.MAP, result.schema().type()); + Schema keySchema = result.schema().keySchema(); + Schema valueSchema = result.schema().valueSchema(); + assertEquals(Type.STRING, keySchema.type()); + assertEquals(Type.INT32, valueSchema.type()); + assertEquals(Time.LOGICAL_NAME, valueSchema.name()); + java.util.Date expected = new SimpleDateFormat(Values.ISO_8601_TIME_FORMAT_PATTERN).parse(timeStr); + assertEquals(Collections.singletonMap(keyStr, expected), result.value()); + } + + @Test + public void shouldParseTimeStringAsTimeInMap() throws Exception { + String keyStr = "k1"; + String timeStr = "14:34:54.346Z"; + String mapStr = "{\"" + keyStr + "\":" + timeStr + "}"; + SchemaAndValue result = Values.parseString(mapStr); + assertEquals(Type.MAP, result.schema().type()); + Schema keySchema = result.schema().keySchema(); + Schema valueSchema = result.schema().valueSchema(); + assertEquals(Type.STRING, keySchema.type()); + assertEquals(Type.INT32, valueSchema.type()); + assertEquals(Time.LOGICAL_NAME, valueSchema.name()); + java.util.Date expected = new SimpleDateFormat(Values.ISO_8601_TIME_FORMAT_PATTERN).parse(timeStr); + assertEquals(Collections.singletonMap(keyStr, expected), result.value()); + } + /** * This is technically invalid JSON, and we don't want to simply ignore the blank elements. */ @@ -432,6 +573,20 @@ public void shouldFailToParseStringOfMapWithIntValuesWithBlankEntries() { Values.convertToMap(Schema.STRING_SCHEMA, " { \"foo\" : \"1234567890\" ,, \"bar\" : \"0\", \"baz\" : \"boz\" } "); } + @Test + public void shouldConsumeMultipleTokens() { + String value = "a:b:c:d:e:f:g:h"; + Parser parser = new Parser(value); + String firstFive = parser.next(5); + assertEquals("a:b:c", firstFive); + assertEquals(":", parser.next()); + assertEquals("d", parser.next()); + assertEquals(":", parser.next()); + String lastEight = parser.next(8); // only 7 remain + assertNull(lastEight); + assertEquals("e", parser.next()); + } + @Test public void shouldParseStringsWithoutDelimiters() { //assertParsed("");