Fix race condition in AppenderatorImpl

[elloooooo]

The method persistAll(Committer committer) in AppenderatorImpl can be call by task-runner thread and publish-driver thread concurrently. There is a chance that the sink.swap() can be called one more time and create an empty FireHydrant and finally leads to a "Trying to persist an empty index!" exception. This leads kafkaIndex task failed. Below is the exception log.

java.util.concurrent.ExecutionException: io.druid.java.util.common.IAE: Trying to persist an empty index!
at com.google.common.util.concurrent.AbstractFuture$Sync.getValue(AbstractFuture.java:299) ~[guava-16.0.1.jar:?]
at com.google.common.util.concurrent.AbstractFuture$Sync.get(AbstractFuture.java:286) ~[guava-16.0.1.jar:?]
at com.google.common.util.concurrent.AbstractFuture.get(AbstractFuture.java:116) ~[guava-16.0.1.jar:?]
at io.druid.indexing.kafka.KafkaIndexTask.lambda$createAndStartPublishExecutor$27(KafkaIndexTask.java:370) ~[?:?]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [?:1.8.0_77]
at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_77]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [?:1.8.0_77]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [?:1.8.0_77]
at java.lang.Thread.run(Thread.java:745) [?:1.8.0_77]
Caused by: io.druid.java.util.common.IAE: Trying to persist an empty index!
at io.druid.segment.IndexMergerV9.persist(IndexMergerV9.java:640) ~[druid-processing-0.12.1-100.jar:0.12.1-100]
at io.druid.segment.IndexMergerV9.persist(IndexMergerV9.java:626) ~[druid-processing-0.12.1-100.jar:0.12.1-100]
at io.druid.segment.realtime.appenderator.AppenderatorImpl.persistHydrant(AppenderatorImpl.java:1185) ~[druid-server-0.12.1-100.jar:0.12.1-100]
at io.druid.segment.realtime.appenderator.AppenderatorImpl.access$300(AppenderatorImpl.java:105) ~[druid-server-0.12.1-100.jar:0.12.1-100]
at io.druid.segment.realtime.appenderator.AppenderatorImpl$2.doCall(AppenderatorImpl.java:441) ~[druid-server-0.12.1-100.jar:0.12.1-100]
at io.druid.common.guava.ThreadRenamingCallable.call(ThreadRenamingCallable.java:44) ~[druid-common-0.12.1-100.jar:0.12.1-100]
... 4 more