Kafka 2140 improve code readability

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

@@ -304,7 +304,7 @@ public static Properties addDeserializerToConfig(Properties properties,
         return newProperties;
     }
 
-    ConsumerConfig(Map<? extends Object, ? extends Object> props) {
+    ConsumerConfig(Map<?, ?> props) {
         super(CONFIG, props);
     }
 

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerRecords.java

@@ -55,7 +55,7 @@ public Iterable<ConsumerRecord<K, V>> records(String topic) {
             throw new IllegalArgumentException("Topic must be non-null.");
         List<List<ConsumerRecord<K, V>>> recs = new ArrayList<List<ConsumerRecord<K, V>>>();
         for (Map.Entry<TopicPartition, List<ConsumerRecord<K, V>>> entry : records.entrySet()) {
-            if (entry.getKey().equals(topic))
+            if (entry.getKey().topic().equals(topic))
                 recs.add(entry.getValue());
         }
         return new ConcatenatedIterable<K, V>(recs);

clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java

@@ -242,7 +242,7 @@ public static Properties addSerializerToConfig(Properties properties,
         return newProperties;
     }
 
-    ProducerConfig(Map<? extends Object, ? extends Object> props) {
+    ProducerConfig(Map<?, ?> props) {
         super(CONFIG, props);
     }
 

clients/src/main/java/org/apache/kafka/common/metrics/stats/Histogram.java

@@ -58,7 +58,7 @@ public void clear() {
 
     @Override
     public String toString() {
-        StringBuilder b = new StringBuilder('{');
+        StringBuilder b = new StringBuilder("{");
         for (int i = 0; i < this.hist.length - 1; i++) {
             b.append(String.format("%.10f", binScheme.fromBin(i)));
             b.append(':');

clients/src/main/java/org/apache/kafka/common/protocol/SecurityProtocol.java

@@ -49,7 +49,7 @@ private SecurityProtocol(int id, String name) {
     }
 
     public static String getName(int id) {
-        return CODE_TO_SECURITY_PROTOCOL.get(id).name;
+        return CODE_TO_SECURITY_PROTOCOL.get((short) id).name;
     }
 
     public static List<String> getNames() {

clients/src/test/java/org/apache/kafka/common/config/AbstractConfigTest.java

@@ -73,7 +73,7 @@ private static class TestConfig extends AbstractConfig {
                                             METRIC_REPORTER_CLASSES_DOC);
         }
 
-        public TestConfig(Map<? extends Object, ? extends Object> props) {
+        public TestConfig(Map<?, ?> props) {
             super(CONFIG, props);
         }
     }

clients/src/test/java/org/apache/kafka/common/protocol/types/ProtocolSerializationTest.java

@@ -43,8 +43,8 @@ public void setup() {
                                  new Field("struct", new Schema(new Field("field", Type.INT32))));
         this.struct = new Struct(this.schema).set("int8", (byte) 1)
                                              .set("int16", (short) 1)
-                                             .set("int32", (int) 1)
-                                             .set("int64", (long) 1)
+                                             .set("int32", 1)
+                                             .set("int64", 1L)
                                              .set("string", "1")
                                              .set("bytes", "1".getBytes())
                                              .set("array", new Object[] {1});

contrib/hadoop-consumer/src/main/java/kafka/etl/KafkaETLContext.java

@@ -159,7 +159,7 @@ public boolean fetchMore () throws IOException {
         _response = _consumer.fetch(fetchRequest);
         if(_response != null) {
             _respIterator = new ArrayList<ByteBufferMessageSet>(){{
-                add((ByteBufferMessageSet) _response.messageSet(_request.getTopic(), _request.getPartition()));
+                add(_response.messageSet(_request.getTopic(), _request.getPartition()));
             }}.iterator();
         }
         _requestTime += (System.currentTimeMillis() - tempTime);

core/src/main/scala/kafka/admin/ReassignPartitionsCommand.scala

@@ -127,7 +127,7 @@ object ReassignPartitionsCommand extends Logging {
     }
     val reassignPartitionsCommand = new ReassignPartitionsCommand(zkClient, partitionsToBeReassigned.toMap)
     // before starting assignment, output the current replica assignment to facilitate rollback
-    val currentPartitionReplicaAssignment = ZkUtils.getReplicaAssignmentForTopics(zkClient, partitionsToBeReassigned.map(_._1.topic).toSeq)
+    val currentPartitionReplicaAssignment = ZkUtils.getReplicaAssignmentForTopics(zkClient, partitionsToBeReassigned.map(_._1.topic))
     println("Current partition replica assignment\n\n%s\n\nSave this to use as the --reassignment-json-file option during rollback"
       .format(ZkUtils.getPartitionReassignmentZkData(currentPartitionReplicaAssignment)))
     // start the reassignment

core/src/main/scala/kafka/admin/TopicCommand.scala

@@ -143,7 +143,7 @@ object TopicCommand {
     topics.foreach { topic =>
       try {
         if (Topic.InternalTopics.contains(topic)) {
-          throw new AdminOperationException("Topic %s is a kafka internal topic and is not allowed to be marked for deletion.".format(topic));
+          throw new AdminOperationException("Topic %s is a kafka internal topic and is not allowed to be marked for deletion.".format(topic))
         } else {
           ZkUtils.createPersistentPath(zkClient, ZkUtils.getDeleteTopicPath(topic))
           println("Topic %s is marked for deletion.".format(topic))

core/src/main/scala/kafka/api/ControlledShutdownRequest.scala

@@ -37,9 +37,9 @@ object ControlledShutdownRequest extends Logging {
   }
 }
 
-case class ControlledShutdownRequest(val versionId: Short,
-                                     val correlationId: Int,
-                                     val brokerId: Int)
+case class ControlledShutdownRequest(versionId: Short,
+                                     correlationId: Int,
+                                     brokerId: Int)
   extends RequestOrResponse(Some(RequestKeys.ControlledShutdownKey)){
 
   def this(correlationId: Int, brokerId: Int) =
@@ -74,4 +74,4 @@ case class ControlledShutdownRequest(val versionId: Short,
     controlledShutdownRequest.append("; BrokerId: " + brokerId)
     controlledShutdownRequest.toString()
   }
-}
+}

core/src/main/scala/kafka/api/ControlledShutdownResponse.scala

@@ -39,9 +39,9 @@ object ControlledShutdownResponse {
 }
 
 
-case class ControlledShutdownResponse(val correlationId: Int,
-                                      val errorCode: Short = ErrorMapping.NoError,
-                                      val partitionsRemaining: Set[TopicAndPartition])
+case class ControlledShutdownResponse(correlationId: Int,
+                                      errorCode: Short = ErrorMapping.NoError,
+                                      partitionsRemaining: Set[TopicAndPartition])
   extends RequestOrResponse() {
   def sizeInBytes(): Int ={
     var size =
@@ -68,4 +68,4 @@ case class ControlledShutdownResponse(val correlationId: Int,
 
   override def describe(details: Boolean):String = { toString }
 
-}
+}

core/src/main/scala/kafka/api/GenericRequestAndHeader.scala

@@ -36,7 +36,7 @@ private[kafka] abstract class GenericRequestAndHeader(val versionId: Short,
     2 /* version id */ +
     4 /* correlation id */ +
     (2 + clientId.length) /* client id */ +
-    body.sizeOf();
+    body.sizeOf()
   }
 
   override def toString(): String = {
@@ -52,4 +52,4 @@ private[kafka] abstract class GenericRequestAndHeader(val versionId: Short,
     strBuffer.append("; Body: " + body.toString)
     strBuffer.toString()
   }
-}
+}

core/src/main/scala/kafka/api/GenericResponseAndHeader.scala

@@ -29,7 +29,7 @@ private[kafka] abstract class GenericResponseAndHeader(val correlationId: Int,
 
   def sizeInBytes(): Int = {
     4 /* correlation id */ +
-    body.sizeOf();
+    body.sizeOf()
   }
 
   override def toString(): String = {
@@ -43,4 +43,4 @@ private[kafka] abstract class GenericResponseAndHeader(val correlationId: Int,
     strBuffer.append("; Body: " + body.toString)
     strBuffer.toString()
   }
-}
+}

core/src/main/scala/kafka/api/LeaderAndIsrRequest.scala

@@ -59,8 +59,8 @@ object PartitionStateInfo {
   }
 }
 
-case class PartitionStateInfo(val leaderIsrAndControllerEpoch: LeaderIsrAndControllerEpoch,
-                              val allReplicas: Set[Int]) {
+case class PartitionStateInfo(leaderIsrAndControllerEpoch: LeaderIsrAndControllerEpoch,
+                              allReplicas: Set[Int]) {
   def replicationFactor = allReplicas.size
 
   def writeTo(buffer: ByteBuffer) {
@@ -200,4 +200,4 @@ case class LeaderAndIsrRequest (versionId: Short,
       leaderAndIsrRequest.append(";PartitionState:" + partitionStateInfos.mkString(","))
     leaderAndIsrRequest.toString()
   }
-}
+}

core/src/main/scala/kafka/api/StopReplicaResponse.scala

@@ -42,9 +42,9 @@ object StopReplicaResponse {
 }
 
 
-case class StopReplicaResponse(val correlationId: Int,
-                               val responseMap: Map[TopicAndPartition, Short],
-                               val errorCode: Short = ErrorMapping.NoError)
+case class StopReplicaResponse(correlationId: Int,
+                               responseMap: Map[TopicAndPartition, Short],
+                               errorCode: Short = ErrorMapping.NoError)
     extends RequestOrResponse() {
   def sizeInBytes(): Int ={
     var size =
@@ -72,4 +72,4 @@ case class StopReplicaResponse(val correlationId: Int,
   }
 
   override def describe(details: Boolean):String = { toString }
-}
+}

core/src/main/scala/kafka/api/TopicMetadata.scala

@@ -109,7 +109,7 @@ object PartitionMetadata {
 }
 
 case class PartitionMetadata(partitionId: Int, 
-                             val leader: Option[BrokerEndPoint],
+                             leader: Option[BrokerEndPoint],
                              replicas: Seq[BrokerEndPoint],
                              isr: Seq[BrokerEndPoint] = Seq.empty,
                              errorCode: Short = ErrorMapping.NoError) extends Logging {

core/src/main/scala/kafka/api/TopicMetadataRequest.scala

@@ -46,10 +46,10 @@ object TopicMetadataRequest extends Logging {
   }
 }
 
-case class TopicMetadataRequest(val versionId: Short,
-                                val correlationId: Int,
-                                val clientId: String,
-                                val topics: Seq[String])
+case class TopicMetadataRequest(versionId: Short,
+                                correlationId: Int,
+                                clientId: String,
+                                topics: Seq[String])
  extends RequestOrResponse(Some(RequestKeys.MetadataKey)){
 
   def this(topics: Seq[String], correlationId: Int) =
@@ -93,4 +93,4 @@ case class TopicMetadataRequest(val versionId: Short,
       topicMetadataRequest.append("; Topics: " + topics.mkString(","))
     topicMetadataRequest.toString()
   }
-}
+}

core/src/main/scala/kafka/client/ClientUtils.scala

@@ -74,7 +74,7 @@ object ClientUtils extends Logging{
     } else {
       debug("Successfully fetched metadata for %d topic(s) %s".format(topics.size, topics))
     }
-    return topicMetadataResponse
+    topicMetadataResponse
   }
 
   /**

core/src/main/scala/kafka/cluster/EndPoint.scala

@@ -68,7 +68,7 @@ case class EndPoint(host: String, port: Int, protocolType: SecurityProtocol) {
   def writeTo(buffer: ByteBuffer): Unit = {
     buffer.putInt(port)
     writeShortString(buffer, host)
-    buffer.putShort(protocolType.id.toShort)
+    buffer.putShort(protocolType.id)
   }
 
   def sizeInBytes: Int =

core/src/main/scala/kafka/controller/ControllerChannelManager.scala

@@ -309,8 +309,8 @@ class ControllerBrokerRequestBatch(controller: KafkaController) extends  Logging
     }
     updateMetadataRequestMap.clear()
     stopReplicaRequestMap foreach { case(broker, replicaInfoList) =>
-      val stopReplicaWithDelete = replicaInfoList.filter(p => p.deletePartition == true).map(i => i.replica).toSet
-      val stopReplicaWithoutDelete = replicaInfoList.filter(p => p.deletePartition == false).map(i => i.replica).toSet
+      val stopReplicaWithDelete = replicaInfoList.filter(_.deletePartition).map(_.replica).toSet
+      val stopReplicaWithoutDelete = replicaInfoList.filterNot(_.deletePartition).map(_.replica).toSet
       debug("The stop replica request (delete = true) sent to broker %d is %s"
         .format(broker, stopReplicaWithDelete.mkString(",")))
       debug("The stop replica request (delete = false) sent to broker %d is %s"

core/src/main/scala/kafka/controller/KafkaController.scala

@@ -647,7 +647,7 @@ class KafkaController(val config : KafkaConfig, zkClient: ZkClient, val brokerSt
    */
   def startup() = {
     inLock(controllerContext.controllerLock) {
-      info("Controller starting up");
+      info("Controller starting up")
       registerSessionExpirationListener()
       isRunning = true
       controllerElector.startup
@@ -1326,7 +1326,7 @@ case class PartitionAndReplica(topic: String, partition: Int, replica: Int) {
   }
 }
 
-case class LeaderIsrAndControllerEpoch(val leaderAndIsr: LeaderAndIsr, controllerEpoch: Int) {
+case class LeaderIsrAndControllerEpoch(leaderAndIsr: LeaderAndIsr, controllerEpoch: Int) {
   override def toString(): String = {
     val leaderAndIsrInfo = new StringBuilder
     leaderAndIsrInfo.append("(Leader:" + leaderAndIsr.leader)

core/src/main/scala/kafka/controller/ReplicaStateMachine.scala

@@ -282,7 +282,7 @@ class ReplicaStateMachine(controller: KafkaController) extends Logging {
     val replicasForTopic = controller.controllerContext.replicasForTopic(topic)
     val replicaStatesForTopic = replicasForTopic.map(r => (r, replicaState(r))).toMap
     debug("Are all replicas for topic %s deleted %s".format(topic, replicaStatesForTopic))
-    replicaStatesForTopic.foldLeft(true)((deletionState, r) => deletionState && r._2 == ReplicaDeletionSuccessful)
+    replicaStatesForTopic.forall(_._2 == ReplicaDeletionSuccessful)
   }
 
   def isAtLeastOneReplicaInDeletionStartedState(topic: String): Boolean = {

core/src/main/scala/kafka/coordinator/DelayedRebalance.scala

@@ -41,8 +41,7 @@ class DelayedRebalance(sessionTimeout: Long,
 
   /* check if all known consumers have requested to re-join group */
   override def tryComplete(): Boolean = {
-    allConsumersJoinedGroup.set(groupRegistry.memberRegistries.values.foldLeft
-      (true) ((agg, cur) => agg && cur.joinGroupReceived.get()))
+    allConsumersJoinedGroup.set(groupRegistry.memberRegistries.values.forall(_.joinGroupReceived.get()))
 
     if (allConsumersJoinedGroup.get())
       forceComplete()

core/src/main/scala/kafka/javaapi/Implicits.scala

@@ -47,9 +47,4 @@ private[javaapi] object Implicits extends Logging {
     }
   }
 
-  // used explicitly by ByteBufferMessageSet constructor as due to SI-4141 which affects Scala 2.8.1, implicits are not visible in constructors
-  implicit def javaListToScalaBuffer[A](l: java.util.List[A]) = {
-    import scala.collection.JavaConversions._
-    l: collection.mutable.Buffer[A]
-  }
 }

core/src/main/scala/kafka/javaapi/message/ByteBufferMessageSet.scala

@@ -19,14 +19,14 @@ package kafka.javaapi.message
 import java.util.concurrent.atomic.AtomicLong
 import java.nio.ByteBuffer
 import kafka.message._
-import kafka.javaapi.Implicits.javaListToScalaBuffer
+
+import scala.collection.JavaConverters._
 
 class ByteBufferMessageSet(val buffer: ByteBuffer) extends MessageSet {
   private val underlying: kafka.message.ByteBufferMessageSet = new kafka.message.ByteBufferMessageSet(buffer)
 
   def this(compressionCodec: CompressionCodec, messages: java.util.List[Message]) {
-    // due to SI-4141 which affects Scala 2.8.1, implicits are not visible in constructors and must be used explicitly
-    this(new kafka.message.ByteBufferMessageSet(compressionCodec, new AtomicLong(0), javaListToScalaBuffer(messages).toSeq : _*).buffer)
+    this(new kafka.message.ByteBufferMessageSet(compressionCodec, new AtomicLong(0), messages.asScala: _*).buffer)
   }
 
   def this(messages: java.util.List[Message]) {

core/src/main/scala/kafka/log/CleanerConfig.scala

@@ -29,13 +29,13 @@ package kafka.log
  * @param enableCleaner Allows completely disabling the log cleaner
  * @param hashAlgorithm The hash algorithm to use in key comparison.
  */
-case class CleanerConfig(val numThreads: Int = 1, 
-                         val dedupeBufferSize: Long = 4*1024*1024L,
-                         val dedupeBufferLoadFactor: Double = 0.9d,
-                         val ioBufferSize: Int = 1024*1024,
-                         val maxMessageSize: Int = 32*1024*1024,
-                         val maxIoBytesPerSecond: Double = Double.MaxValue,
-                         val backOffMs: Long = 15 * 1000,
-                         val enableCleaner: Boolean = true,
-                         val hashAlgorithm: String = "MD5") {
-}
+case class CleanerConfig(numThreads: Int = 1,
+                         dedupeBufferSize: Long = 4*1024*1024L,
+                         dedupeBufferLoadFactor: Double = 0.9d,
+                         ioBufferSize: Int = 1024*1024,
+                         maxMessageSize: Int = 32*1024*1024,
+                         maxIoBytesPerSecond: Double = Double.MaxValue,
+                         backOffMs: Long = 15 * 1000,
+                         enableCleaner: Boolean = true,
+                         hashAlgorithm: String = "MD5") {
+}

core/src/main/scala/kafka/log/LogCleaner.scala

@@ -72,7 +72,7 @@ class LogCleaner(val config: CleanerConfig,
                  time: Time = SystemTime) extends Logging with KafkaMetricsGroup {
 
   /* for managing the state of partitions being cleaned. package-private to allow access in tests */
-  private[log] val cleanerManager = new LogCleanerManager(logDirs, logs);
+  private[log] val cleanerManager = new LogCleanerManager(logDirs, logs)
 
   /* a throttle used to limit the I/O of all the cleaner threads to a user-specified maximum rate */
   private val throttler = new Throttler(desiredRatePerSec = config.maxIoBytesPerSecond, 
@@ -622,4 +622,4 @@ private case class LogToClean(topicPartition: TopicAndPartition, log: Log, first
   val cleanableRatio = dirtyBytes / totalBytes.toDouble
   def totalBytes = cleanBytes + dirtyBytes
   override def compare(that: LogToClean): Int = math.signum(this.cleanableRatio - that.cleanableRatio).toInt
-}
+}