KAFKA-8457: Move log from replica into partition

[soondenana]

A partition object contain one or many replica objects. These replica
objects in turn can have the "log" if the replica corresponds to the
local node. All the code in Partition or ReplicaManager peek into
replica object to fetch the log if they need to operate on that. As
replica object can represent a local replica or a remote one, this
lead to a bunch of "if-else" code in log fetch and offset update code.

NOTE: In addition to a "log" that is in use during normal operation, if
an alter log directory command is issued, we also create a future log
object. This object catches up with local log and then we switch the log
directory. So temporarily a Partition can have two local logs. Before
this change both logs are inside replica objects.

This change is an attempt to untangle this relationship. In particular
it moves "log" from a replica object to Partition. So a partition contains
a local log to which all writes go. And it maintains a list of replica
for offset and "caught up time" data that it uses for replication
protocol. The replica correspoding to Local node contains a log object,
but the object is now read only and no code except Replica and test code
use it. Every other part of code in Partion and ReplicaManger use the
log object stored in Partition. This uncouples the replica-log relation
and all the "if-else" code went away. Couple of more structural changes
are made in this change:

1. Two subclasses of Replica are introduced: LocalReplica and
   RemoteReplica. This makes it clear what each replica stores and is
   capable of.
2. The "log" in Partition is also wrapped in a LogInfo wrapper, which
   encapuslates all the code that either operated on "log" or maintained
   state of it.

Unit tests have been updated to take care of change in heirarchy.
Tested by running multiple brokers and produced and consumed data. Also
changed log directory back and forth to make sure that alter log
directory use case works.

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]

retest this please

JDK 11 passed, JDK 8 failed with some configuration error:

18:40:16 ERROR: Step ?Publish JUnit test result report? failed: No test report files were found. Configuration error?

[hachikuji]

@soondenana Thanks for the PR. I had two high level comments, but this is looking promising!

[jsancio]

Thanks for the PR. Much simpler data model.

[soondenana]

Thanks for the PR. Much simpler data model.

Hi Jose, took care of all comments except making allReplicaMap containing only RemoteReplica. Please take a look at my response and let me know what you think.

[hachikuji]

Thanks, nice cleanup. Left a few comments.

[hachikuji]

A few compiler errors:

17:42:16 /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk8-scala2.11/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala:898: missing parameter type
17:42:16         .thenAnswer(_ => {
17:42:16                     ^
17:42:24 /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk8-scala2.11/core/src/test/scala/unit/kafka/server/ReplicaAlterLogDirsThreadTest.scala:421: type mismatch;
17:42:24  found   : () => Unit
17:42:24  required: org.easymock.IAnswer[_ <: Unit]
17:42:24       .andAnswer(() => {
17:42:24                     ^
17:42:24 /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk8-scala2.11/core/src/test/scala/unit/kafka/server/ReplicaAlterLogDirsThreadTest.scala:668: type mismatch;
17:42:24  found   : () => Unit
17:42:24  required: org.easymock.IAnswer[_ <: Unit]
17:42:24       .andAnswer(() => {
17:42:24                     ^

[soondenana]

A few compiler errors:

17:42:16 /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk8-scala2.11/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala:898: missing parameter type
17:42:16         .thenAnswer(_ => {
17:42:16                     ^
17:42:24 /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk8-scala2.11/core/src/test/scala/unit/kafka/server/ReplicaAlterLogDirsThreadTest.scala:421: type mismatch;
17:42:24  found   : () => Unit
17:42:24  required: org.easymock.IAnswer[_ <: Unit]
17:42:24       .andAnswer(() => {
17:42:24                     ^
17:42:24 /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk8-scala2.11/core/src/test/scala/unit/kafka/server/ReplicaAlterLogDirsThreadTest.scala:668: type mismatch;
17:42:24  found   : () => Unit
17:42:24  required: org.easymock.IAnswer[_ <: Unit]
17:42:24       .andAnswer(() => {
17:42:24                     ^

Yup, going through Java8/2.11 compilation errors. Took out some dead code, but Scala 2.11 doesn't like it. I guess I need to downgrade my local setup to this version.

[jsancio]

Thanks for the changes. This is a good improvement. Just left some minor comments.

[hachikuji]

Thanks, just a few more comments.

[hachikuji]

LGTM. Thanks for the patch!