fix: change default exception handling for timestamp extractors

[agavra]

fixes #2609

Description

ksqlDB currently fails violently whenever a timestamp cannot be extracted from a record. Perhaps more fatally, it also does not log anything to the processing log meaning users that are using the cloud offering of ksql will not be able to see what happened to their queries.

This PR addresses both of those concerns and gives users the ability to toggle whether they want the failure to be violent or graceful so that they can choose not to lose data.

Testing done

New unit tests and end to end testing:

ksql> PRINT ts FROM BEGINNING;
Value format: JSON or KAFKA_STRING
rowtime: 2/25/20 3:11:47 PM PST, key: <null>, value: {"TS":"10-04-19 12:00:17"}
rowtime: 2/25/20 3:11:52 PM PST, key: <null>, value: {"TS":"10-04-19 X 12:00:17"}
rowtime: 2/25/20 3:11:55 PM PST, key: <null>, value: {"TS":"10-04-19 12:00:17"}

ksql> CREATE STREAM ts (ts VARCHAR) WITH (value_format='json', kafka_topic='ts', partitions=1, timestamp='ts', timestamp_format='yy-MM-dd HH:mm:ss')

ksql> SELECT * FROM ts EMIT CHANGES;
+--------------------------------------------------------------+--------------------------------------------------------------+--------------------------------------------------------------+
|ROWTIME                                                       |ROWKEY                                                        |TS                                                            |
+--------------------------------------------------------------+--------------------------------------------------------------+--------------------------------------------------------------+
|1271703617000                                                 |null                                                          |10-04-19 12:00:17                                             |
|1271703617000                                                 |null                                                          |10-04-19 12:00:17                                             |
^CQuery terminated

ksql> SELECT * FROM KSQL_PROCESSING_LOG EMIT CHANGES;
+--------------+--------------+--------------+--------------+--------------+--------------+
|ROWTIME       |ROWKEY        |LOGGER        |LEVEL         |TIME          |MESSAGE       |
+--------------+--------------+--------------+--------------+--------------+--------------+
|1582672325285 |null          |processing.211|ERROR         |1582672325277 |{TYPE=0, DESER|
|              |              |78117353354788|              |              |IALIZATIONERRO|
|              |              |25            |              |              |R={ERRORMESSAG|
|              |              |              |              |              |E=Text '10-04-|
|              |              |              |              |              |19 X 12:00:17'|
|              |              |              |              |              | could not be |
|              |              |              |              |              |parsed at inde|
|              |              |              |              |              |x 9, RECORDB64|
|              |              |              |              |              |=null, CAUSE=[|
|              |              |              |              |              |]}, RECORDPROC|
|              |              |              |              |              |ESSINGERROR=nu|
|              |              |              |              |              |ll, PRODUCTION|
|              |              |              |              |              |ERROR=null}   |

Reviewer checklist

• Ensure docs are updated if necessary. (eg. if a user visible feature is being added or changed).
• Ensure relevant issues are linked (description should include text like "Fixes #")

[spena]

Will this allow KSQL to display the row with ROWTIME = -1 in case of a failure?

I'm trying to understand how the FailOnInvalidTimestamp will work in this situation. If a -1 is found on the row, then the FailOnInvalidTimestamp will throw an exception which this code will catch and log it to the processing log. But it returns a -1 back to stream, so the stream will set it to as ROWTIME, won't it?

[agavra]

any negative timestamp causes the event to be ignored by Kafka Streams. That means that if this wraps FailOnInvalidTimestamp but you set ksql.timestamp.throw.on.invalid to false then it will suppress the throw and just ignore the message

[rodesai]

I'd use a different message type for this error. Deserialization error messages generally mean the serde couldn't deserialize the record in the topic, and contain a base64 encoded representation of the data inside (in a field called recordB64). The data written out here represents something different - KSQL's view of the row once it's been deserialized.

[rodesai]

You should be able to use the existing record processing error (EngineProcessingLogMessageFactory.recordProcessingError)

[agavra]

i was debating which I should do since this felt like a type of deserialization error to me, but I'll update it to a new type (I can see both sides)

[JimGalasyn]

Nice!

[rodesai]

LGTM!