[KAFKA-2804] Create / Update changelog topics upon state store initialization

• Type: Sub-task
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 0.10.0.0
• Assignee: Guozhang Wang
• Reporter: Guozhang Wang
• Created: 11/Nov/15 00:21
• Updated: 07/Dec/15 23:25
• Resolved: 07/Dec/15 23:25

Description

When state store instances that are logging-backed are initialized, we need to check if the corresponding change log topics have been created with the right number of partitions:

1) If not exist, create topic
2) If expected #.partitions < actual #.partitions, delete and re-create topic.
3) If expected #.partitions > actual #.partitions, add partitions.