[LI-HOTFIX] Prevent ReplicaFetcherThread from throwing UnknownTopicOr…

…PartitionException upon topic creation and deletion. (apache#80)

TICKET =
LI_DESCRIPTION =
When does UnknownTopicOrPartitionException typically occur?
* Upon a topic creation, a follower broker of a new partition starts replica fetcher before the prospective leader broker of the new partition receives the leadership information from the controller. Apache Kafka has a an open issue about this (see KAFKA-6221)
* Upon a topic deletion, a follower broker of a to-be-deleted partition starts replica fetcher after the leader broker of the to-be-deleted partition processes the deletion information from the controller.
* As expected, clusters with frequent topic creation and deletion report UnknownTopicOrPartitionException with relatively higher frequency.

What is the impact?
* Exception tracking systems identify the error logs with UnknownTopicOrPartitionException as an exception. This results in a lot of noise for a transient issue that is expected to recover by itself and a natural process in Kafka due to its asynchronous state propagation.

Why not move it to a lower than warn-level log?
* Despite typically being a transient issue, UnknownTopicOrPartitionException may also indicate real issues if it doesn't fix itself after a short period of time. To ensure detection of such scenarios, this PR sets the log level to warn.

EXIT_CRITERIA = TICKET [KAFKA-6221]

core/src/main/scala/kafka/server/AbstractFetcherThread.scala

@@ -369,6 +369,11 @@ abstract class AbstractFetcherThread(name: String,
                     "that the partition is being moved")
                   partitionsWithError += topicPartition
 
+                case Errors.UNKNOWN_TOPIC_OR_PARTITION =>
+                  warn(s"Remote broker does not host the partition $topicPartition, which could indicate " +
+                    "that the partition is being created or deleted.")
+                  partitionsWithError += topicPartition
+
                 case _ =>
                   error(s"Error for partition $topicPartition at offset ${currentFetchState.fetchOffset}",
                     partitionData.error.exception)