From 817f852781031ef835f692929a06fb0ec3595835 Mon Sep 17 00:00:00 2001 From: Onur Karaman Date: Fri, 24 Apr 2015 13:14:57 -0700 Subject: [PATCH] fix default InterBrokerProtocolVersion --- core/src/main/scala/kafka/server/KafkaConfig.scala | 4 ++-- core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala | 3 ++- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala index cfbbd2b..15e18e0 100755 --- a/core/src/main/scala/kafka/server/KafkaConfig.scala +++ b/core/src/main/scala/kafka/server/KafkaConfig.scala @@ -19,7 +19,7 @@ package kafka.server import java.util.Properties -import kafka.api.ApiVersion +import kafka.api.{KAFKA_082, ApiVersion} import kafka.cluster.EndPoint import kafka.consumer.ConsumerConfig import kafka.message.{BrokerCompressionCodec, CompressionCodec, Message, MessageSet} @@ -104,7 +104,7 @@ object Defaults { val LeaderImbalanceCheckIntervalSeconds = 300 val UncleanLeaderElectionEnable = true val InterBrokerSecurityProtocol = SecurityProtocol.PLAINTEXT.toString - val InterBrokerProtocolVersion = ApiVersion.latestVersion.toString + val InterBrokerProtocolVersion = KAFKA_082.toString /** ********* Controlled shutdown configuration ***********/ val ControlledShutdownMaxRetries = 3 diff --git a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala index 2428dbd..d589ee7 100755 --- a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala +++ b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala @@ -233,10 +233,11 @@ class KafkaConfigTest extends JUnit3Suite { @Test def testVersionConfiguration() { val props = new Properties() + // KIP-2: The idea is that in 0.8.3, we will default wire.protocol.version to 0.8.2 props.put(KafkaConfig.BrokerIdProp, "1") props.put(KafkaConfig.ZkConnectProp, "localhost:2181") val conf = KafkaConfig.fromProps(props) - assertEquals(ApiVersion.latestVersion, conf.interBrokerProtocolVersion) + assertEquals(KAFKA_082, conf.interBrokerProtocolVersion) props.put(KafkaConfig.InterBrokerProtocolVersionProp,"0.8.2.0") val conf2 = KafkaConfig.fromProps(props) -- 1.7.12.4