Split the message storage logic into independent classes

[Demogorgon314]

Motivation

Currently, the KoP store message logic is in KafkaRequestHandler, to avoid duplicate code and easier to support idempotent produce, we must refactor the message storage logic.

Modifications

• Refactor message storage logic.

• The ReplicaManager is a public interface to store messages, we should always use ReplicaManager to store the message in any situation.

• PartitionLog mapping to Kafka is Log, the reason for using this name is to avoid naming conflict, like: log.debug().

• PartitionLogManager hold all PartitionLog, key is full partition name, value is PartitionLog.

TODO

• Add fetch operation

[BewareMyPower]

NOTE: I'll go out for a while and the review is not completed. You can address these comments first.

[BewareMyPower]

LGTM but left some minor comments.

[Hongten]

It seems those two if can combine together. Like

if (ex != null || !isAuthorized) {
    if (ex != null) {
        log.error("Write topic authorize failed, topic - {}. {}",
                fullPartitionName, ex.getMessage());
    }
    unauthorizedTopicResponsesMap.put(topicPartition,
            new ProduceResponse.PartitionResponse(Errors.TOPIC_AUTHORIZATION_FAILED));
    completeOne.run();
    return;
}