From cd4575e0f79bec43f846c02be5b31efa7db66143 Mon Sep 17 00:00:00 2001 From: Lee Dongjin Date: Wed, 15 May 2019 22:27:13 +0100 Subject: [PATCH 1/6] Remove ununsed import from CommandLineUtils.scala --- core/src/main/scala/kafka/utils/CommandLineUtils.scala | 1 - 1 file changed, 1 deletion(-) diff --git a/core/src/main/scala/kafka/utils/CommandLineUtils.scala b/core/src/main/scala/kafka/utils/CommandLineUtils.scala index 1bf7cdf78e8ac..b19da3ec15271 100644 --- a/core/src/main/scala/kafka/utils/CommandLineUtils.scala +++ b/core/src/main/scala/kafka/utils/CommandLineUtils.scala @@ -19,7 +19,6 @@ import java.util.Properties import joptsimple.{OptionParser, OptionSet, OptionSpec} -import org.apache.kafka.common.utils.AppInfoParser import scala.collection.Set From 724ccf678ff4d3bcd77851cd9bc38905daad6537 Mon Sep 17 00:00:00 2001 From: Lee Dongjin Date: Mon, 20 May 2019 06:36:53 +0900 Subject: [PATCH 2/6] Remove unused imports from CommandLineUtils.scala, KafkaConfig.scala --- core/src/main/scala/kafka/server/KafkaConfig.scala | 2 +- core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala | 1 - 2 files changed, 1 insertion(+), 2 deletions(-) diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala index 584620fa370cf..60af03c186e6b 100755 --- a/core/src/main/scala/kafka/server/KafkaConfig.scala +++ b/core/src/main/scala/kafka/server/KafkaConfig.scala @@ -29,7 +29,7 @@ import kafka.utils.CoreUtils import kafka.utils.Implicits._ import org.apache.kafka.clients.CommonClientConfigs import org.apache.kafka.common.Reconfigurable -import org.apache.kafka.common.config.ConfigDef.{ConfigKey, ValidList, Validator} +import org.apache.kafka.common.config.ConfigDef.{ConfigKey, ValidList} import org.apache.kafka.common.config.internals.BrokerSecurityConfigs import org.apache.kafka.common.config.{AbstractConfig, ConfigDef, ConfigException, SaslConfigs, SslClientAuth, SslConfigs, TopicConfig} import org.apache.kafka.common.metrics.Sensor diff --git a/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala b/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala index ddcee122510ac..5192e65ae4df7 100644 --- a/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala +++ b/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala @@ -14,7 +14,6 @@ package kafka.server -import java.nio.ByteBuffer import java.util.{Collections, LinkedHashMap, Optional, Properties} import java.util.concurrent.{Executors, Future, TimeUnit} From d3ad1185a8f5f367934f0df87a61c0cf422b7ee8 Mon Sep 17 00:00:00 2001 From: Lee Dongjin Date: Mon, 20 May 2019 06:46:15 +0900 Subject: [PATCH 3/6] Add ignore case in KafkaController#process, GroupCoordinator#doSyncGroup, and GroupCoordinator#handleHeartbeat. --- core/src/main/scala/kafka/controller/KafkaController.scala | 1 + .../main/scala/kafka/coordinator/group/GroupCoordinator.scala | 4 ++++ 2 files changed, 5 insertions(+) diff --git a/core/src/main/scala/kafka/controller/KafkaController.scala b/core/src/main/scala/kafka/controller/KafkaController.scala index 2034c74971d68..ac377bff06bf6 100644 --- a/core/src/main/scala/kafka/controller/KafkaController.scala +++ b/core/src/main/scala/kafka/controller/KafkaController.scala @@ -1602,6 +1602,7 @@ class KafkaController(val config: KafkaConfig, processIsrChangeNotification() case Startup => processStartup() + case _ => } } catch { case e: ControllerMovedException => diff --git a/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala b/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala index 55c90e56ae1d8..fac1fdbd1596d 100644 --- a/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala +++ b/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala @@ -401,6 +401,8 @@ class GroupCoordinator(val brokerId: Int, val memberMetadata = group.get(memberId) responseCallback(memberMetadata.assignment, Errors.NONE) completeAndScheduleNextHeartbeatExpiration(group, group.get(memberId)) + + case _ => } } } @@ -533,6 +535,8 @@ class GroupCoordinator(val brokerId: Int, val member = group.get(memberId) completeAndScheduleNextHeartbeatExpiration(group, member) responseCallback(Errors.NONE) + + case _ => } } } From a2ad4f65737c0b38d80e779a958cb1ffe74cfca1 Mon Sep 17 00:00:00 2001 From: Lee Dongjin Date: Wed, 22 May 2019 12:49:07 +0900 Subject: [PATCH 4/6] Fix blind pattern matching in GroupCoordinator to explicit one --- .../scala/kafka/coordinator/group/GroupCoordinator.scala | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala b/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala index fac1fdbd1596d..5c08d94a47b26 100644 --- a/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala +++ b/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala @@ -402,7 +402,9 @@ class GroupCoordinator(val brokerId: Int, responseCallback(memberMetadata.assignment, Errors.NONE) completeAndScheduleNextHeartbeatExpiration(group, group.get(memberId)) - case _ => + case Dead => + // this case is already handled, so raise exception + throw new IllegalStateException("can't reach this line") } } } @@ -536,7 +538,9 @@ class GroupCoordinator(val brokerId: Int, completeAndScheduleNextHeartbeatExpiration(group, member) responseCallback(Errors.NONE) - case _ => + case Dead => + // this case is already handled, so raise exception + throw new IllegalStateException("can't reach this line") } } } From 37606da5639c45027742ee00b238762f3a29c042 Mon Sep 17 00:00:00 2001 From: Lee Dongjin Date: Fri, 24 May 2019 09:37:45 +0900 Subject: [PATCH 5/6] Change Exception message for GroupCoordinator#[doSyncGroup, handleHeartbeat] --- .../scala/kafka/coordinator/group/GroupCoordinator.scala | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala b/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala index 5c08d94a47b26..daf38f35da14d 100644 --- a/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala +++ b/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala @@ -403,8 +403,7 @@ class GroupCoordinator(val brokerId: Int, completeAndScheduleNextHeartbeatExpiration(group, group.get(memberId)) case Dead => - // this case is already handled, so raise exception - throw new IllegalStateException("can't reach this line") + throw new IllegalStateException(s"Reached unexpected condition for Dead group ${group.groupId}") } } } @@ -539,8 +538,7 @@ class GroupCoordinator(val brokerId: Int, responseCallback(Errors.NONE) case Dead => - // this case is already handled, so raise exception - throw new IllegalStateException("can't reach this line") + throw new IllegalStateException(s"Reached unexpected condition for Dead group $groupId") } } } From 54e32f8ead515b4f8f82f25d5f63c0b53b3c45ed Mon Sep 17 00:00:00 2001 From: Lee Dongjin Date: Fri, 24 May 2019 09:38:24 +0900 Subject: [PATCH 6/6] Fix unexhaustive match case in KafkaController#process --- core/src/main/scala/kafka/controller/KafkaController.scala | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/core/src/main/scala/kafka/controller/KafkaController.scala b/core/src/main/scala/kafka/controller/KafkaController.scala index ac377bff06bf6..0880967a601bd 100644 --- a/core/src/main/scala/kafka/controller/KafkaController.scala +++ b/core/src/main/scala/kafka/controller/KafkaController.scala @@ -1602,7 +1602,8 @@ class KafkaController(val config: KafkaConfig, processIsrChangeNotification() case Startup => processStartup() - case _ => + case ShutdownEventThread => + // not handled here } } catch { case e: ControllerMovedException =>