SAMZA-2431: Fix the checkpoint and changelog topic auto-creation.#1251
Conversation
|
@prateekm Can you please take a look at this patch. |
shanthoosh
left a comment
shanthoosh
left a comment
There was a problem hiding this comment.
Thanks for the review.
| checkpointTopicProperties.putAll(spec.getConfig()); | ||
| kafkaSpec = KafkaStreamSpec.fromSpec(StreamSpec.createCheckpointStreamSpec(spec.getPhysicalName(), spec.getSystemName())) | ||
| .copyWithReplicationFactor(Integer.parseInt(new KafkaConfig(config).getCheckpointReplicationFactor().get())) | ||
| .copyWithProperties(checkpointTopicProperties); |
There was a problem hiding this comment.
We can do it separately, can we add .copyWithConfig or .copyWithMap which takes Config or Map<String, String> ?
copyWithProperties is not user friendly, as we are converting a Map/Config to a properties, and this method basically convert it back.
There was a problem hiding this comment.
Yes, adding the copyWithConfig to KafkaStreamSpec might be good convenience API. We've additional cleanup planned with this control-flow in the near-future. If acceptable, then I can couple this change with that. Since it's a critical bug-fix, I would prefer to keep the change minimal by containing it to the fix alone and do the clean-up later.
…ache#1251) * Fix the checkpoint and changelog topic creation configurations. * Address review comments. * Address review comments.
3 participants
Symptom: Checkpoint and changelog kafka topics of a samza job may be created with cleanup.policy set to 'delete' instead of 'compact' for certain cases.
Cause:
Changes: Fix the topic-creation control-flow for the metadata topics and generate the correct topic-configurations.
Tests: Added unit tests to validate that the expected topic configuration bag was generated for both checkpoint and changelog topics.
API Changes: None
Upgrade Instructions: None
Usage Instructions: None