...

When partition reassignment removes topic partitions from a offline broker, those removed partitions become orphan partitions to the broker. When the offline broker comes back online, it is not able to clean up both data and folders that belong to orphan partitions.  Log manager will scan all log dirs during startup, but the time based retention policy on a topic partition will not be kicked out until the replicaHighWatermark  has been set. replicaHighWatermark of a partition is set when the broker becomes either leader or follower of the partition.  Orphan partitions will never have chance to get replicaHighWatermark set.  In addition, we do not have logic to delete folders that belong to orphan partition today.  This KIP provides a mechanism for brokers to remove orphan partitions automatically. 

Public Interfaces

• No change in public Interface.  This feature will be turned on automatically.Add two metrics:  1) kafka.log:type=LogManager,name=OrphanLogPartitionCount  (the number of orphan partitions) and 2) kafka.log:type=LogManager,name=OrphanLogPartitionSize (the size of orphan partitions)

Proposed Changes

a) Provide a mechanism to remove orphan partition automatically. 

...