Fixed the regression for stream-table join. Table should always read … #495
Conversation
…from the begining of the topic.
|
When was the regression introduced? what's the effect? That the join is empty? Something else? |
|
This also seems like a good time to add a unit test :) |
|
The regression was introduced when we were refactoring the code. When we join stream-table, the table always should be read from the beginning otherwise the join will return null values for the right hand side. |
|
retest this please |
| final SchemaKStream schemaKStream = right.buildStream(builder, ksqlConfig, kafkaTopicClient, metastoreUtil, functionRegistry, props); | ||
| Map<String, Object> joinTableProps = new HashMap<>(); | ||
| joinTableProps.putAll(props); | ||
| joinTableProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "EARLIEST"); |
There was a problem hiding this comment.
should "EARLIEST" be "earliest"?
There was a problem hiding this comment.
Yes, but in KSQL it's case insensitive. Regardless, I think it's better to use the correct one. Changed it to earliest :)
| final SchemaKStream schemaKStream = right.buildStream(builder, ksqlConfig, kafkaTopicClient, metastoreUtil, functionRegistry, props); | ||
| Map<String, Object> joinTableProps = new HashMap<>(); | ||
| joinTableProps.putAll(props); | ||
| joinTableProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "EARLIEST"); |
There was a problem hiding this comment.
probably be better to expose this as a KSQLConfig property - and default it to EARLIEST. There might be some edge cases where tables are massive and cannot be bootstrapped etc
There was a problem hiding this comment.
auto.offset.reset is configurable for KSQL, however, this one is a special case for Stream-Table join and is required to always be earliest and if we have it as latest the join would be incorrect.
There was a problem hiding this comment.
Thats what I mean - this is a special case and we dont always know what people will do with the system. It is possible that there will be an instance where the hardcoded 'earliest' is not wanted. exposing a new property 'table.auto.offset.reset' would work around this.
There was a problem hiding this comment.
I see what you mean but in this case earliest is the only correct option and setting it to anything else will result in incorrect query results. So even if we have table.auto.offset.reset it's value cannot be changed and should always be earliest even when we have the table change topic containing huge number of rows.
| ("EARLIEST")) { | ||
| return mock(SchemaKTable.class); | ||
| } else { | ||
| throw new KsqlException("auto.offset.reset should be set to EARLIEST."); |
There was a problem hiding this comment.
So if this regresses again, this exception will be thrown in the test?
There was a problem hiding this comment.
Yes, if the Stream-table join changes the table auto.offset.reset this will catch it.
…from the beginning of the topic.