-
Notifications
You must be signed in to change notification settings - Fork 3.7k
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.
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
Remove kafka lookup records when a record is tombstoned #12819
Remove kafka lookup records when a record is tombstoned #12819
Conversation
Travis didn't trigger for some reason. I closed and re-opened the PR. changes LGTM. I will merge once CI passes. |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Great fix, thanks!
Should we document this behavior in the docs? The only reference I could find is here, but that doesn't address streaming updates.
The docs are actually buried in the extensions area. I've updated them to mention this new behaviour. |
|
||
long start = System.currentTimeMillis(); | ||
while (events == factory.getCompletedEventCount()) { | ||
Thread.sleep(100); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
why not use 10
here as well?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thanks @abhishekagarwal87 I've pushed that change, sorry for the delay.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Looks like it fails on CI with a lower sleep time. It was tested locally a few times and passed with the lower value but I will revert the change so the CI can pass. Never mind it looks like an unrelated test caused the failure
thank you. I will merge this once CI is green. |
Description
Lookups made using the
druid-kafka-extraction-namespace
extension do not currently handle records that have been deleted from a kafka topic. Deleting a record from a kafka topic involves sending a record with a null payload and the same key as the record you want to delete to the topic. This is also known as sending a tombstone. This PR aims to handle key value records from kafka that have keys but also null values.On receiving a key value record that has a key and a null value, the corresponding record stored in the factory map will be removed.
Key changed/added classes in this PR
KafkaLookupExtractorFactory.java
This PR has: