KAFKA-8813: Refresh log config if it's updated before initialization

[soondenana]

A partition log in initialized in following steps:

1. Fetch log config from ZK
2. Call LogManager.getOrCreateLog which creates the Log object, then
3. Registers the Log object

Step #3 enables Configuration update thread to deliver configuration
updates to the log. But if any update arrives between step #1 and #3
then that update is missed. It breaks following use case:

1. Create a topic with default configuration, and immediately after that
2. Update the configuration of topic

There is a race condition here and in random cases update made in
seocond step will get dropped.

This change fixes it by tracking updates arriving between step #1 and #3
Once a Partition is done initialzing log, it checks if it has missed any
update. If yes, then the configuration is read from ZK again.

Added unit tests to make sure a dirty configuration is refreshed. Tested
on local cluster to make sure that topic configuration and updates are
handled correctly.

More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.

Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[soondenana]

cc: @hachikuji, @jsancio, @mumrah

Maybe useful to review the change by hiding whitespace changes (put ?w=1 to the end of url or use the UI)

[soondenana]

Ran create topic, change config in a loop (ran 100 times) and made sure that configs update correctly. Here is the bash command I used to create and change topic:

for idx in `seq 100`; do 
    bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 3 --partitions 3 --topic test-topic-${idx}
    bin/kafka-configs.sh --alter --entity-type topics --entity-name test-topic-${idx} --add-config 'min.insync.replicas=2' --zookeeper localhost:2181
done

Here is one to check the configuration:

for idx in `seq 100`; do 
    bin/kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic test-topic-${idx}
done 2> /dev/null | grep min.insync

Here is to clean things up at the end:

for idx in `seq 100`; do 
    bin/kafka-topics.sh --delete --bootstrap-server localhost:9092  --topic test-topic-${idx}
done

[soondenana]

Tested again with 500 topic creation followed by update (instead of 100 in previous comment). Checked to make sure topic configuration is sane.

[hachikuji]

Thanks for the patch. Left a couple initial comments.

[hachikuji]

Sorry for the delay. Left a few more comments.

[hachikuji]

Just a couple more comments.

[hachikuji]

LGTM. Thanks for the patch.

[soondenana]

retest this please