Skip to content

KAFKA-9074: Correct Connect’s Values.parseString to properly parse a time and timestamp literal #7568

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rhauch merged 5 commits into from
Feb 4, 2020
Merged

KAFKA-9074: Correct Connect’s Values.parseString to properly parse a time and timestamp literal #7568

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
d6a5243
KAFKA-9074: Correct Connect’s `Values.parseString` to properly parse …
rhauch Oct 21, 2019
39922d8
Fixed checkstyle problems
rhauch Oct 21, 2019
e22892d
Incorporated feedback
rhauch Oct 22, 2019
435b15f
KAFKA-9074: Corrected the parsing of logical types within arrays and …
rhauch Oct 23, 2019
f53e431
Rebased on updated logic in Values class
rhauch Jan 22, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
133 changes: 102 additions & 31 deletions connect/api/src/main/java/org/apache/kafka/connect/data/Values.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -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<String> 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 = ",";
Expand Down Expand Up @@ -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);
Expand All @@ -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);
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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);
}
}

Expand Down Expand Up @@ -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;
Expand All @@ -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);
}

Expand All @@ -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 ...
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -1088,6 +1145,7 @@ public int mark() {
public void rewindTo(int position) {
iter.setIndex(position);
nextToken = null;
previousToken = null;
}

public String original() {
Expand All @@ -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) {
Comment thread
rhauch marked this conversation as resolved.
Outdated
if (!hasNext()) {
rewindTo(start);
return null;
}
next();
}
return original.substring(current, position());
}

private String consumeNextToken() throws NoSuchElementException {
boolean escaped = false;
int start = iter.getIndex();
Expand Down
Loading