[KSQL] rekey example failing after ksqdb version upgrade

[gAmUssA]

during the test via the harness. the output from actual ksql is different from the expected output.
key format should be KAFKA_INT

related #495

[gAmUssA]

@colinhicks welcome back. could you take a look on this?

[colinhicks]

We had diagnosed this before:
#482 (comment)

This test doesn't always fail under CI. If data is inserted fast enough, KSQL can infer the more specific key format before the PRINT command is executed. If it's slower, KSQL has less info about the format specificity.

One way to address this is to introduce an optional delay between commands sent by the harness. In this case, this would allow the PRINT command to be deferred so that the reported key format is stable across CI runs.

[big-andy-coates]

The ambiguity comes about because the first key(s) successfully deserialize as both KAFKA_INT and KAFKA_STRING. When it processes later rows they fail to deserialize as KAFKA_STRING and hence we end up with KAFKA_INT.

PRINT will poll Kafka and iterate over the entire first batch of messages to try and determine the key and value formats. It then outputs its best guess and the rows in the first batch, before polling again. If any candidate formats are excluded while processing a batch, PRINT will output the new, reduced, candidate list and reprocess the batch.

As Colin alludes to, the timing seems to be the cause of the flakiness here. It depends what comes back in the first batch as to what the output is.

Taking the example output from #482 (comment):

Key format: KAFKA_INT or KAFKA_STRING
Value format: AVRO
rowtime: 2020/07/23 18:00:37.348 Z, key: 780, value: {"RATING": 2.1}
Key format: KAFKA_INT
rowtime: 2020/07/23 18:00:36.359 Z, key: 294, value: {"RATING": 8.2}
rowtime: 2020/07/23 18:00:36.697 Z, key: 354, value: {"RATING": 9.9}
rowtime: 2020/07/23 18:00:37.152 Z, key: 128, value: {"RATING": 8.7}
rowtime: 2020/07/23 18:00:36.591 Z, key: 294, value: {"RATING": 8.5}
rowtime: 2020/07/23 18:00:36.821 Z, key: 354, value: {"RATING": 9.7}
rowtime: 2020/07/23 18:00:36.950 Z, key: 782, value: {"RATING": 7.8}
rowtime: 2020/07/23 18:00:37.057 Z, key: 782, value: {"RATING": 7.7}
rowtime: 2020/07/23 18:00:37.237 Z, key: 128, value: {"RATING": 8.4}
Topic printing ceased

I would read this that the first bath only contains one row:

rowtime: 2020/07/23 18:00:37.348 Z, key: 780, value: {"RATING": 2.1}

The key in this row wasn't enough to exclude KAFKA_STRING from the set of candidates. However, something in the latter rows, (it's not clear how many were in the next batch), excluded KAFKA_STRING.

The way to fix this is to ensure the first row output has a key that can't be deserialized as KAFKA_STRING, i.e. determine which of these rows:

rowtime: 2020/07/23 18:00:36.359 Z, key: 294, value: {"RATING": 8.2}
rowtime: 2020/07/23 18:00:36.697 Z, key: 354, value: {"RATING": 9.9}
rowtime: 2020/07/23 18:00:37.152 Z, key: 128, value: {"RATING": 8.7}
rowtime: 2020/07/23 18:00:36.591 Z, key: 294, value: {"RATING": 8.5}
rowtime: 2020/07/23 18:00:36.821 Z, key: 354, value: {"RATING": 9.7}
rowtime: 2020/07/23 18:00:36.950 Z, key: 782, value: {"RATING": 7.8}
rowtime: 2020/07/23 18:00:37.057 Z, key: 782, value: {"RATING": 7.7}
rowtime: 2020/07/23 18:00:37.237 Z, key: 128, value: {"RATING": 8.4}

Causes KAFKA_STRING to be excluded and make that the first row in the output.