You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/01/29 19:10:29 UTC
[GitHub] [kafka] rondagostino opened a new pull request #10005: Add ConfigRepository, use in Partition
rondagostino opened a new pull request #10005:
URL: https://github.com/apache/kafka/pull/10005
`Partition` objects are able to retrieve topic configs when creating their log, and currently they do so via ZooKeeper. ZooKeeper is not available when using the Raft-based metadata log, so we need to abstract the retrieval of configs so it can work either with or without ZooKeeper. This PR introduces `trait ConfigRepository` with `ZkConfigRepository` and `LocalConfigRepository` implementations. `ReplicaManager` holds the instance, and `Partition` gets it from there.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rondagostino commented on a change in pull request #10005: MINOR: Add ConfigRepository, use in Partition and KafkaApis
Posted by GitBox <gi...@apache.org>.
rondagostino commented on a change in pull request #10005:
URL: https://github.com/apache/kafka/pull/10005#discussion_r569852775
##########
File path: core/src/main/scala/kafka/cluster/Partition.scala
##########
@@ -89,8 +88,7 @@ object Partition extends KafkaMetricsGroup {
val configProvider = new TopicConfigFetcher {
override def fetch(): Properties = {
- val adminZkClient = new AdminZkClient(replicaManager.zkClient)
- adminZkClient.fetchEntityConfig(ConfigType.Topic, topicPartition.topic)
+ replicaManager.configRepository.topicConfigs(topicPartition.topic())
Review comment:
`TopicConfigFetcher` is specific to a single topic; `ConfigRepository` is not. I don't think one can directly replace the other -- unless I am misunderstanding/missing something?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10005: MINOR: Add ConfigRepository, use in Partition and KafkaApis
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10005:
URL: https://github.com/apache/kafka/pull/10005#discussion_r569780361
##########
File path: core/src/main/scala/kafka/server/metadata/LocalConfigRepository.scala
##########
@@ -0,0 +1,93 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.server.metadata
+
+import java.util
+import java.util.Properties
+import java.util.concurrent.ConcurrentHashMap
+import java.util.function.BiFunction
+
+import org.apache.kafka.common.config.ConfigResource
+import org.apache.kafka.common.config.ConfigResource.Type
+
+import scala.jdk.CollectionConverters._
+
+/**
+ * A ConfigRepository that stores configurations locally.
+ */
+class LocalConfigRepository extends ConfigRepository {
+ val configMap = new ConcurrentHashMap[ConfigResource, util.HashMap[String, String]]
+
+ def setTopicConfig(topic: String, key: String, value: String): Unit = {
+ setConfig(new ConfigResource(Type.TOPIC, topic), key, value)
+ }
+
+ def setBrokerConfig(id: Int, key: String, value: String): Unit = {
+ setConfig(new ConfigResource(Type.BROKER, id.toString()), key, value)
+ }
+
+ def setConfig(configResource: ConfigResource, key: String, value: String): Unit = {
+ configMap.compute(configResource, new BiFunction[ConfigResource, util.HashMap[String, String], util.HashMap[String, String]] {
+ override def apply(resource: ConfigResource,
+ curConfigs: util.HashMap[String, String]): util.HashMap[String, String] = {
+ if (value == null) {
+ if (curConfigs == null) {
+ null
+ } else {
+ val newConfigs = new util.HashMap[String, String](curConfigs)
+ newConfigs.remove(key)
+ if (newConfigs.isEmpty) {
+ null
+ } else {
+ newConfigs
+ }
+ }
+ } else {
+ if (curConfigs == null) {
+ val newConfigs = new util.HashMap[String, String](1)
+ newConfigs.put(key, value)
+ newConfigs
+ } else {
+ val newConfigs = new util.HashMap[String, String](curConfigs.size() + 1)
+ newConfigs.putAll(curConfigs)
+ newConfigs.put(key, value)
+ newConfigs
+ }
+ }
+ }
+ })
+ }
+
+ def config(configResource: ConfigResource): Properties = {
Review comment:
nit: can we be consistent on pluralization? I think I tend to see `config` as plural already, so maybe we can use `topicConfig` and `brokerConfig`.
Really we could even pull this up to `ConfigRepository`. Maybe something like this:
```scala
trait ConfigRepository {
def config(configResource: ConfigResource): Properties
def topicConfig(topic: String): Properties = {
config(new ConfigResource(Type.TOPIC, topic))
}
def brokerConfig(id: Int): Properties = {
config(new ConfigResource(Type.BROKER, id.toString()))
}
}
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rondagostino commented on a change in pull request #10005: Add ConfigRepository, use in Partition and KafkaApis
Posted by GitBox <gi...@apache.org>.
rondagostino commented on a change in pull request #10005:
URL: https://github.com/apache/kafka/pull/10005#discussion_r567313320
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -249,6 +270,7 @@ class KafkaApis(val requestChannel: RequestChannel,
}
def handleLeaderAndIsrRequest(request: RequestChannel.Request): Unit = {
+ rejectWithUnsupportedVersionIfUsingRaftMetadataQuorum(request)
Review comment:
We probably should, yes. When looking at it, the exclusion should be done in `ApiVersion.apiVersionsResponse()`. But how we can know there if we are using a Raft-based metadata quorum is unclear. We could brute-force send in the `KafkaConfig`, but it appears perhaps `Features` might be another way to do it. Have we considered creating a feature for that, do you know?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji merged pull request #10005: MINOR: Add ConfigRepository, use in Partition and KafkaApis
Posted by GitBox <gi...@apache.org>.
hachikuji merged pull request #10005:
URL: https://github.com/apache/kafka/pull/10005
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10005: MINOR: Add ConfigRepository, use in Partition and KafkaApis
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10005:
URL: https://github.com/apache/kafka/pull/10005#discussion_r569770342
##########
File path: core/src/main/scala/kafka/cluster/Partition.scala
##########
@@ -89,8 +88,7 @@ object Partition extends KafkaMetricsGroup {
val configProvider = new TopicConfigFetcher {
override def fetch(): Properties = {
- val adminZkClient = new AdminZkClient(replicaManager.zkClient)
- adminZkClient.fetchEntityConfig(ConfigType.Topic, topicPartition.topic)
+ replicaManager.configRepository.topicConfigs(topicPartition.topic())
Review comment:
Hmm.. It seems like `ConfigRepository` can just replace `TopicConfigFetcher`?
##########
File path: core/src/main/scala/kafka/server/metadata/LocalConfigRepository.scala
##########
@@ -0,0 +1,93 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.server.metadata
+
+import java.util
+import java.util.Properties
+import java.util.concurrent.ConcurrentHashMap
+import java.util.function.BiFunction
+
+import org.apache.kafka.common.config.ConfigResource
+import org.apache.kafka.common.config.ConfigResource.Type
+
+import scala.jdk.CollectionConverters._
+
+/**
+ * A ConfigRepository that stores configurations locally.
+ */
+class LocalConfigRepository extends ConfigRepository {
+ val configMap = new ConcurrentHashMap[ConfigResource, util.HashMap[String, String]]
+
+ def setTopicConfig(topic: String, key: String, value: String): Unit = {
Review comment:
Useful to add some documentation about the contract. For example, it looks like we expect value=null to indicate deletion (and on that note, it might be clearer to have an explicit `removeTopicConfig`).
##########
File path: core/src/main/scala/kafka/server/metadata/ConfigRepository.scala
##########
@@ -0,0 +1,26 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.server.metadata
+
+import java.util.Properties
+
+trait ConfigRepository {
+ def topicConfigs(name: String): Properties
Review comment:
nit: can we be explicit and rename the args to `topicName` and `brokerId`?
Also, it would be useful to document the fact that the returned value is a copy of current configuration and will not reflect future changes.
##########
File path: core/src/main/scala/kafka/server/metadata/LocalConfigRepository.scala
##########
@@ -0,0 +1,93 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.server.metadata
+
+import java.util
+import java.util.Properties
+import java.util.concurrent.ConcurrentHashMap
+import java.util.function.BiFunction
+
+import org.apache.kafka.common.config.ConfigResource
+import org.apache.kafka.common.config.ConfigResource.Type
+
+import scala.jdk.CollectionConverters._
+
+/**
+ * A ConfigRepository that stores configurations locally.
+ */
+class LocalConfigRepository extends ConfigRepository {
+ val configMap = new ConcurrentHashMap[ConfigResource, util.HashMap[String, String]]
+
+ def setTopicConfig(topic: String, key: String, value: String): Unit = {
+ setConfig(new ConfigResource(Type.TOPIC, topic), key, value)
+ }
+
+ def setBrokerConfig(id: Int, key: String, value: String): Unit = {
+ setConfig(new ConfigResource(Type.BROKER, id.toString()), key, value)
+ }
+
+ def setConfig(configResource: ConfigResource, key: String, value: String): Unit = {
+ configMap.compute(configResource, new BiFunction[ConfigResource, util.HashMap[String, String], util.HashMap[String, String]] {
+ override def apply(resource: ConfigResource,
+ curConfigs: util.HashMap[String, String]): util.HashMap[String, String] = {
+ if (value == null) {
+ if (curConfigs == null) {
+ null
+ } else {
+ val newConfigs = new util.HashMap[String, String](curConfigs)
+ newConfigs.remove(key)
+ if (newConfigs.isEmpty) {
+ null
+ } else {
+ newConfigs
+ }
+ }
+ } else {
+ if (curConfigs == null) {
+ val newConfigs = new util.HashMap[String, String](1)
+ newConfigs.put(key, value)
+ newConfigs
+ } else {
+ val newConfigs = new util.HashMap[String, String](curConfigs.size() + 1)
+ newConfigs.putAll(curConfigs)
+ newConfigs.put(key, value)
+ newConfigs
+ }
+ }
+ }
+ })
+ }
+
+ def config(configResource: ConfigResource): Properties = {
Review comment:
nit: can we be consistent on pluralization? I think I tend to see `config` as plural already, so maybe we can use `topicConfig` and `brokerConfig`.
Really we could even pull this up to `ConfigRepository`. Maybe something like this:
```scala
trait ConfigRepository {
def config(configResource: ConfigResource): Properties
def topicConfig(topic: String): Properties = {
config(new ConfigResource(Type.TOPIC, topic))
}
override def brokerConfig(id: Int): Properties = {
config(new ConfigResource(Type.BROKER, id.toString()))
}
}
```
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -120,6 +120,7 @@ class KafkaApis(val requestChannel: RequestChannel,
type FetchResponseStats = Map[TopicPartition, RecordConversionStats]
this.logIdent = "[KafkaApi-%d] ".format(brokerId)
val adminZkClient = new AdminZkClient(zkClient)
+ val configHelper = new ConfigHelper(metadataCache, config, replicaManager.configRepository)
Review comment:
nit: it's a little ugly to reach into `ReplicaManager` in order to get the dependency we need. This also makes the mocking more difficult. Can we instead add `ConfigRepository` as a formal parameter?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rondagostino commented on a change in pull request #10005: Add ConfigRepository, use in Partition and KafkaApis
Posted by GitBox <gi...@apache.org>.
rondagostino commented on a change in pull request #10005:
URL: https://github.com/apache/kafka/pull/10005#discussion_r567300660
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -1654,6 +1702,7 @@ class KafkaApis(val requestChannel: RequestChannel,
.format(response, request.header.correlationId, request.header.clientId))
response
}
+
Review comment:
I'm not sure how all of these spacing changes happened -- I didn't explicitly do them. My guess is the editor tried to be helpful by formatting things at some point. I reverted these.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rondagostino commented on a change in pull request #10005: MINOR: Add ConfigRepository, use in Partition and KafkaApis
Posted by GitBox <gi...@apache.org>.
rondagostino commented on a change in pull request #10005:
URL: https://github.com/apache/kafka/pull/10005#discussion_r570326339
##########
File path: core/src/main/scala/kafka/server/KafkaServer.scala
##########
@@ -348,15 +348,15 @@ class KafkaServer(
/* start processing requests */
dataPlaneRequestProcessor = new KafkaApis(socketServer.dataPlaneRequestChannel, replicaManager, adminManager, groupCoordinator, transactionCoordinator,
- kafkaController, forwardingManager, zkClient, config.brokerId, config, metadataCache, metrics, authorizer, quotaManagers,
+ kafkaController, forwardingManager, zkClient, config.brokerId, config, replicaManager.configRepository, metadataCache, metrics, authorizer, quotaManagers,
Review comment:
@hachikuji After this change we have only 1 place left where we reach into `ReplicaManager` to grab its `ConfigRepository` -- when we create a `Partition` instance. Do you think we should change that so that we explicitly provide the instance to the `Partition` constructor and the companion object's `apply()` method? Aside form this last question I think we're all set here.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji edited a comment on pull request #10005: MINOR: Add ConfigRepository, use in Partition and KafkaApis
Posted by GitBox <gi...@apache.org>.
hachikuji edited a comment on pull request #10005:
URL: https://github.com/apache/kafka/pull/10005#issuecomment-772130047
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10005: Add ConfigRepository, use in Partition and KafkaApis
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10005:
URL: https://github.com/apache/kafka/pull/10005#discussion_r567202218
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -119,7 +122,21 @@ class KafkaApis(val requestChannel: RequestChannel,
type FetchResponseStats = Map[TopicPartition, RecordConversionStats]
this.logIdent = "[KafkaApi-%d] ".format(brokerId)
- val adminZkClient = new AdminZkClient(zkClient)
+
+ val usingRaftMetadataQuorum = if (adminManager != null && controller != null && zkClient != null) {
Review comment:
Can we use `config.requiresZookeeper` here?
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -3316,8 +3440,8 @@ class KafkaApis(val requestChannel: RequestChannel,
}
private def parseForwardedClientAddress(
- address: Array[Byte]
- ): InetAddress = {
+ address: Array[Byte]
Review comment:
nit: fix indentation
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -119,7 +122,21 @@ class KafkaApis(val requestChannel: RequestChannel,
type FetchResponseStats = Map[TopicPartition, RecordConversionStats]
this.logIdent = "[KafkaApi-%d] ".format(brokerId)
- val adminZkClient = new AdminZkClient(zkClient)
+
+ val usingRaftMetadataQuorum = if (adminManager != null && controller != null && zkClient != null) {
+ false
+ } else { // at least one is null, so make sure they are all null, plus we must have a forwardingManager
+ if (adminManager != null || controller != null || zkClient != null) {
Review comment:
This kind of logic might be better suited for a factory. Something like this:
```scala
object KafkaApis {
def buildZk(zkClient: KafkaZkClient, ...): KafkaApis = {
// verify adminManager, controller, etc
}
def buildSelfManaged(...): KafkaApis = {
// verify forwarding manager, etc.
}
}
```
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -150,7 +167,11 @@ class KafkaApis(val requestChannel: RequestChannel,
}
forwardingManager match {
- case Some(mgr) if !request.isForwarded && !controller.isActive =>
+ case Some(mgr) if request.isForwarded && usingRaftMetadataQuorum =>
Review comment:
The extra checks in here seem unnecessary. If the forwarding manager is provided, we can use it.
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -1654,6 +1702,7 @@ class KafkaApis(val requestChannel: RequestChannel,
.format(response, request.header.correlationId, request.header.clientId))
response
}
+
Review comment:
nit: do we need all these additional newlines? they are a tad distracting
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -2609,14 +2672,27 @@ class KafkaApis(val requestChannel: RequestChannel,
case rt => throw new InvalidRequestException(s"Unexpected resource type $rt")
}
}
- val authorizedResult = adminManager.alterConfigs(authorizedResources, alterConfigsRequest.validateOnly)
- val unauthorizedResult = unauthorizedResources.keys.map { resource =>
- resource -> configsAuthorizationApiError(resource)
+ // Should never get here when using a Raft metadata quorum since it would be forwarded,
Review comment:
If you think the protection is worthwhile, then can we do something like `rejectWithUnsupportedVersionIfUsingRaftMetadataQuorum`?
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -3008,7 +3107,8 @@ class KafkaApis(val requestChannel: RequestChannel,
})
}
- if (!authHelper.authorize(request.context, ALTER, CLUSTER, CLUSTER_NAME)) {
+ // not yet supported when using a Raft-based metadata quorum
+ if (usingRaftMetadataQuorum || !authHelper.authorize(request.context, ALTER, CLUSTER, CLUSTER_NAME)) {
Review comment:
`CLUSTER_AUTHORIZATION_FAILED` seems like a surprising error for an unsupported operation. Any of the following errors might be better:
- `UNKNOWN_SERVER_ERROR`
- `OPERATION_NOT_ATTEMPTED`
- `UNSUPPORTE_VERSION`
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -2965,10 +3064,10 @@ class KafkaApis(val requestChannel: RequestChannel,
val electionRequest = request.body[ElectLeadersRequest]
def sendResponseCallback(
- error: ApiError
- )(
- results: Map[TopicPartition, ApiError]
- ): Unit = {
+ error: ApiError
Review comment:
nit: not sure what happened with the indentation here
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -3450,7 +3574,187 @@ class KafkaApis(val requestChannel: RequestChannel,
brokerEpochInRequest < controller.brokerEpoch
}
}
+}
+
+class ConfigHelper(metadataCache: MetadataCache, config: KafkaConfig, configRepository: ConfigRepository) extends Logging {
Review comment:
Can we move this to a separate file?
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -249,6 +270,7 @@ class KafkaApis(val requestChannel: RequestChannel,
}
def handleLeaderAndIsrRequest(request: RequestChannel.Request): Unit = {
+ rejectWithUnsupportedVersionIfUsingRaftMetadataQuorum(request)
Review comment:
Do we excludes these APIs from the ApiVersions response when raft is enabled?
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -97,12 +100,12 @@ import scala.annotation.nowarn
*/
class KafkaApis(val requestChannel: RequestChannel,
val replicaManager: ReplicaManager,
- val adminManager: ZkAdminManager,
+ adminManager: ZkAdminManager,
val groupCoordinator: GroupCoordinator,
val txnCoordinator: TransactionCoordinator,
- val controller: KafkaController,
- val forwardingManager: Option[ForwardingManager],
- val zkClient: KafkaZkClient,
+ controller: KafkaController,
+ forwardingManager: Option[ForwardingManager],
+ zkClient: KafkaZkClient,
Review comment:
Can we use `Option` for all of these types that might be null?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10005: MINOR: Add ConfigRepository, use in Partition and KafkaApis
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10005:
URL: https://github.com/apache/kafka/pull/10005#discussion_r570404954
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -1881,7 +1884,11 @@ class ReplicaManager(val config: KafkaConfig,
logManager.handleLogDirFailure(dir)
if (sendZkNotification)
- zkClient.propagateLogDirEvent(localBrokerId)
+ if (zkClient == null) {
Review comment:
Can we make `zkClient` an `Option`? By the way, we are really close to being able to remove `KafkaZkClient` as a dependency, but I guess we will first need #9577.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10005: MINOR: Add ConfigRepository, use in Partition and KafkaApis
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10005:
URL: https://github.com/apache/kafka/pull/10005#discussion_r570404954
##########
File path: core/src/main/scala/kafka/server/ReplicaManager.scala
##########
@@ -1881,7 +1884,11 @@ class ReplicaManager(val config: KafkaConfig,
logManager.handleLogDirFailure(dir)
if (sendZkNotification)
- zkClient.propagateLogDirEvent(localBrokerId)
+ if (zkClient == null) {
Review comment:
Can we make `zkClient` an `Option`? By the way, we are really close to being able to remove `KafkaZkClient` as a dependency, but I guess we will first need #9577.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rondagostino commented on a change in pull request #10005: MINOR: Add ConfigRepository, use in Partition and KafkaApis
Posted by GitBox <gi...@apache.org>.
rondagostino commented on a change in pull request #10005:
URL: https://github.com/apache/kafka/pull/10005#discussion_r570326339
##########
File path: core/src/main/scala/kafka/server/KafkaServer.scala
##########
@@ -348,15 +348,15 @@ class KafkaServer(
/* start processing requests */
dataPlaneRequestProcessor = new KafkaApis(socketServer.dataPlaneRequestChannel, replicaManager, adminManager, groupCoordinator, transactionCoordinator,
- kafkaController, forwardingManager, zkClient, config.brokerId, config, metadataCache, metrics, authorizer, quotaManagers,
+ kafkaController, forwardingManager, zkClient, config.brokerId, config, replicaManager.configRepository, metadataCache, metrics, authorizer, quotaManagers,
Review comment:
@hachikuji After this change we have only 1 place left where we reach into `ReplicaManager` to grab its `ConfigRepository` -- when we create a `Partition` instance. Do you think we should change that so that we explicitly provide the instance to the `Partition` constructor and the companion object's `apply()` method? Aside form this last question I think we're all set here.
##########
File path: core/src/main/scala/kafka/server/KafkaServer.scala
##########
@@ -348,15 +348,15 @@ class KafkaServer(
/* start processing requests */
dataPlaneRequestProcessor = new KafkaApis(socketServer.dataPlaneRequestChannel, replicaManager, adminManager, groupCoordinator, transactionCoordinator,
- kafkaController, forwardingManager, zkClient, config.brokerId, config, metadataCache, metrics, authorizer, quotaManagers,
+ kafkaController, forwardingManager, zkClient, config.brokerId, config, replicaManager.configRepository, metadataCache, metrics, authorizer, quotaManagers,
Review comment:
It's just the `apply()` method that reaches in, so I'm going to add an explicit `ConfigRepository` parameter there.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10005: Add ConfigRepository, use in Partition and KafkaApis
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10005:
URL: https://github.com/apache/kafka/pull/10005#discussion_r567201967
##########
File path: core/src/main/scala/kafka/log/LogConfig.scala
##########
@@ -374,40 +374,4 @@ object LogConfig {
MessageTimestampDifferenceMaxMsProp -> KafkaConfig.LogMessageTimestampDifferenceMaxMsProp,
MessageDownConversionEnableProp -> KafkaConfig.LogMessageDownConversionEnableProp
)
-
-
- /**
- * Copy the subset of properties that are relevant to Logs. The individual properties
Review comment:
We moved this here from `KafkaServer`. Not sure why we want to revert that. `LogConfig` seems like the right place to have a function to extract the log config properties.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10005: MINOR: Add ConfigRepository, use in Partition and KafkaApis
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10005:
URL: https://github.com/apache/kafka/pull/10005#discussion_r569854419
##########
File path: core/src/main/scala/kafka/cluster/Partition.scala
##########
@@ -89,8 +88,7 @@ object Partition extends KafkaMetricsGroup {
val configProvider = new TopicConfigFetcher {
override def fetch(): Properties = {
- val adminZkClient = new AdminZkClient(replicaManager.zkClient)
- adminZkClient.fetchEntityConfig(ConfigType.Topic, topicPartition.topic)
+ replicaManager.configRepository.topicConfigs(topicPartition.topic())
Review comment:
We have access to the topic name in `Partition`, so my thought was to replace the calls to `TopicConfigFetcher.fetch` with `ConfigRepository.topicConfig(topic)`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rondagostino commented on a change in pull request #10005: Add ConfigRepository, use in Partition and KafkaApis
Posted by GitBox <gi...@apache.org>.
rondagostino commented on a change in pull request #10005:
URL: https://github.com/apache/kafka/pull/10005#discussion_r567300510
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -2609,14 +2672,27 @@ class KafkaApis(val requestChannel: RequestChannel,
case rt => throw new InvalidRequestException(s"Unexpected resource type $rt")
}
}
- val authorizedResult = adminManager.alterConfigs(authorizedResources, alterConfigsRequest.validateOnly)
- val unauthorizedResult = unauthorizedResources.keys.map { resource =>
- resource -> configsAuthorizationApiError(resource)
+ // Should never get here when using a Raft metadata quorum since it would be forwarded,
Review comment:
I think `NOT_CONTROLLER` is the clearest thing we could return.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rondagostino commented on a change in pull request #10005: Add ConfigRepository, use in Partition and KafkaApis
Posted by GitBox <gi...@apache.org>.
rondagostino commented on a change in pull request #10005:
URL: https://github.com/apache/kafka/pull/10005#discussion_r567315872
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -3008,7 +3107,8 @@ class KafkaApis(val requestChannel: RequestChannel,
})
}
- if (!authHelper.authorize(request.context, ALTER, CLUSTER, CLUSTER_NAME)) {
+ // not yet supported when using a Raft-based metadata quorum
+ if (usingRaftMetadataQuorum || !authHelper.authorize(request.context, ALTER, CLUSTER, CLUSTER_NAME)) {
Review comment:
Changed to `UnsupportedVersionException`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rondagostino commented on a change in pull request #10005: MINOR: Add ConfigRepository, use in Partition and KafkaApis
Posted by GitBox <gi...@apache.org>.
rondagostino commented on a change in pull request #10005:
URL: https://github.com/apache/kafka/pull/10005#discussion_r569715123
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -249,6 +270,7 @@ class KafkaApis(val requestChannel: RequestChannel,
}
def handleLeaderAndIsrRequest(request: RequestChannel.Request): Unit = {
+ rejectWithUnsupportedVersionIfUsingRaftMetadataQuorum(request)
Review comment:
This no longer applies to this PR as we have removed the `KafkaApis` changes from it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rondagostino commented on a change in pull request #10005: Add ConfigRepository, use in Partition and KafkaApis
Posted by GitBox <gi...@apache.org>.
rondagostino commented on a change in pull request #10005:
URL: https://github.com/apache/kafka/pull/10005#discussion_r567300461
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -150,7 +167,11 @@ class KafkaApis(val requestChannel: RequestChannel,
}
forwardingManager match {
- case Some(mgr) if !request.isForwarded && !controller.isActive =>
+ case Some(mgr) if request.isForwarded && usingRaftMetadataQuorum =>
Review comment:
I reworked it a bit to clarify the intent, which is to add a log message if a Raft-based broker receives a forwarded request; it'll still handle the request (and we'll give `NOT_CONTROLLER`).
I'm not sure we should change the logic here and forward a request that itself is already forwarded as that opens up the possibility of endless loops.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10005: MINOR: Add ConfigRepository, use in Partition and KafkaApis
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10005:
URL: https://github.com/apache/kafka/pull/10005#discussion_r569944146
##########
File path: core/src/main/scala/kafka/server/metadata/LocalConfigRepository.scala
##########
@@ -0,0 +1,136 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.server.metadata
+
+import java.util
+import java.util.Properties
+import java.util.concurrent.ConcurrentHashMap
+import java.util.function.BiFunction
+
+import org.apache.kafka.common.config.ConfigResource
+import org.apache.kafka.common.config.ConfigResource.Type
+
+import scala.jdk.CollectionConverters._
+
+/**
+ * A ConfigRepository that stores configurations locally.
+ */
+class LocalConfigRepository extends ConfigRepository {
+ val configMap = new ConcurrentHashMap[ConfigResource, util.HashMap[String, String]]
+
+ /**
+ * Remove the topic config for the given topic name and the given key.
+ *
+ * @param topicName the name of the topic for which the config will be removed
+ * @param key the key identifying the topic config to remove
+ */
+ def removeTopicConfig(topicName: String, key: String): Unit = {
+ setTopicConfig(topicName, key, null)
+ }
+
+ /**
+ * Set the topic config for the given topic name and the given key to the given value.
+ *
+ * @param topicName the name of the topic for which the config will be set
+ * @param key the key identifying the topic config to set
+ * @param value the value to set for the topic config with null implying a removal
+ */
+ def setTopicConfig(topicName: String, key: String, value: String): Unit = {
+ setConfig(new ConfigResource(Type.TOPIC, topicName), key, value)
+ }
+
+ /**
+ * Remove the broker config for the given broker ID and the given key.
+ *
+ * @param brokerId the ID of the broker for which the config will be removed
+ * @param key the key identifying the topic config to remove
+ */
+ def removeBrokerConfig(brokerId: Int, key: String): Unit = {
Review comment:
nit: I think it is better to have just one way to remove configs. I know I suggested this, but I was assuming we would not also support removal through `set`. If you think using `set` with a null value is clear enough, we can remove these.
##########
File path: core/src/main/scala/kafka/cluster/Partition.scala
##########
@@ -342,7 +331,7 @@ class Partition(val topicPartition: TopicPartition,
// Visible for testing
private[cluster] def createLog(isNew: Boolean, isFutureReplica: Boolean, offsetCheckpoints: OffsetCheckpoints): Log = {
def fetchLogConfig: LogConfig = {
- val props = topicConfigProvider.fetch()
+ val props = configRepository.topicConfig(topicPartition.topic())
Review comment:
nit: we can use `configRepository.topicConfig(topic)`
##########
File path: core/src/main/scala/kafka/server/KafkaServer.scala
##########
@@ -348,15 +348,15 @@ class KafkaServer(
/* start processing requests */
dataPlaneRequestProcessor = new KafkaApis(socketServer.dataPlaneRequestChannel, replicaManager, adminManager, groupCoordinator, transactionCoordinator,
- kafkaController, forwardingManager, zkClient, config.brokerId, config, metadataCache, metrics, authorizer, quotaManagers,
+ kafkaController, forwardingManager, zkClient, config.brokerId, config, replicaManager.configRepository, metadataCache, metrics, authorizer, quotaManagers,
Review comment:
nit: Similarly here, it would be cleaner to construct the config repository separately in this class and pass it down to the components that need it including `ReplicaManager` and `KafkaApis`.
##########
File path: core/src/main/scala/kafka/server/metadata/ZkConfigRepository.scala
##########
@@ -0,0 +1,39 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.server.metadata
+
+import java.util.Properties
+
+import kafka.server.ConfigType
+import kafka.zk.{AdminZkClient, KafkaZkClient}
+import org.apache.kafka.common.config.ConfigResource
+import org.apache.kafka.common.config.ConfigResource.Type
+
+
+object ZkConfigRepository {
+ def apply(zkClient: KafkaZkClient): ZkConfigRepository =
+ new ZkConfigRepository(new AdminZkClient(zkClient))
+}
+
+class ZkConfigRepository(adminZkClient: AdminZkClient) extends ConfigRepository {
+ override def config(configResource: ConfigResource): Properties = {
+ adminZkClient.fetchEntityConfig(
+ if (configResource.`type`() == Type.TOPIC) ConfigType.Topic else ConfigType.Broker,
Review comment:
Hmm... It would be better to have an explicit check for the `Broker` type. We can raise an exception for any remaining types.
##########
File path: core/src/main/scala/kafka/server/metadata/LocalConfigRepository.scala
##########
@@ -0,0 +1,136 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.server.metadata
+
+import java.util
+import java.util.Properties
+import java.util.concurrent.ConcurrentHashMap
+import java.util.function.BiFunction
+
+import org.apache.kafka.common.config.ConfigResource
+import org.apache.kafka.common.config.ConfigResource.Type
+
+import scala.jdk.CollectionConverters._
+
+/**
+ * A ConfigRepository that stores configurations locally.
+ */
+class LocalConfigRepository extends ConfigRepository {
Review comment:
I found the term "local" a little confusing. How about `CachedConfigRepository`?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] rondagostino commented on a change in pull request #10005: MINOR: Add ConfigRepository, use in Partition and KafkaApis
Posted by GitBox <gi...@apache.org>.
rondagostino commented on a change in pull request #10005:
URL: https://github.com/apache/kafka/pull/10005#discussion_r570342092
##########
File path: core/src/main/scala/kafka/server/KafkaServer.scala
##########
@@ -348,15 +348,15 @@ class KafkaServer(
/* start processing requests */
dataPlaneRequestProcessor = new KafkaApis(socketServer.dataPlaneRequestChannel, replicaManager, adminManager, groupCoordinator, transactionCoordinator,
- kafkaController, forwardingManager, zkClient, config.brokerId, config, metadataCache, metrics, authorizer, quotaManagers,
+ kafkaController, forwardingManager, zkClient, config.brokerId, config, replicaManager.configRepository, metadataCache, metrics, authorizer, quotaManagers,
Review comment:
It's just the `apply()` method that reaches in, so I'm going to add an explicit `ConfigRepository` parameter there.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10005: MINOR: Add ConfigRepository, use in Partition and KafkaApis
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10005:
URL: https://github.com/apache/kafka/pull/10005#discussion_r569854419
##########
File path: core/src/main/scala/kafka/cluster/Partition.scala
##########
@@ -89,8 +88,7 @@ object Partition extends KafkaMetricsGroup {
val configProvider = new TopicConfigFetcher {
override def fetch(): Properties = {
- val adminZkClient = new AdminZkClient(replicaManager.zkClient)
- adminZkClient.fetchEntityConfig(ConfigType.Topic, topicPartition.topic)
+ replicaManager.configRepository.topicConfigs(topicPartition.topic())
Review comment:
We have access to the topic name in `Partition`, so my thought was to replace the calls to `TopicConfigFetcher.fetch()` with `ConfigRepository.topicConfig(topic)`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji merged pull request #10005: MINOR: Add ConfigRepository, use in Partition and KafkaApis
Posted by GitBox <gi...@apache.org>.
hachikuji merged pull request #10005:
URL: https://github.com/apache/kafka/pull/10005
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #10005: Add ConfigRepository, use in Partition and KafkaApis
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10005:
URL: https://github.com/apache/kafka/pull/10005#discussion_r567203060
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -249,6 +270,7 @@ class KafkaApis(val requestChannel: RequestChannel,
}
def handleLeaderAndIsrRequest(request: RequestChannel.Request): Unit = {
+ rejectWithUnsupportedVersionIfUsingRaftMetadataQuorum(request)
Review comment:
Do we excludes these APIs from the ApiVersions response when raft is enabled? It would be a little weird if `ApiVersions` indicates that an API is supported, but then we return UNSUPPORTED_VERSION when the client tries to use it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on pull request #10005: MINOR: Add ConfigRepository, use in Partition and KafkaApis
Posted by GitBox <gi...@apache.org>.
hachikuji commented on pull request #10005:
URL: https://github.com/apache/kafka/pull/10005#issuecomment-772130047
@rondagostino Thanks for the updates. I am wondering if we can pull out the changes in `KafkaApis` since most of them are unrelated to the `ConfigRepository` refactor. In general, I think we need a better approach for managing the dependency requirements for the zk/self-managed implementations. One idea here is try to encapsulate the zk-specific dependencies into a separate object. For example:
```scala
trait MetadataSupport
class ZkSupport(
adminManager: ZkAdminManager,
zkClient: KafkaZkClient,
...
) extends MetadataSupport
```
Then we can pass through `MetadataSupport` and get rid of the optional types. We can then have helpers like the following:
```scala
def requireZk(): ZkSupport = {
support match {
case zkSupport: ZkSupport => zkSupport
case _ => throw new IllegalStateException
}
}
```
The advantage is that we can make the dependencies that are available when zk support is enabled explicit. We don't have to deal with the permutations that are possible when we have 5 different `Option` classes.
We can then start making this trait more useful by adding methods. For example, instead of reaching into the `KafkaController` to get the broker epoch, we can add a `brokerEpoch` method. Eventually we might get to a point where we don't need so many special cases.
The second thing I think we can do better is how we handle the acceptable scope of the APIs when using zk or the self-managed quorum. The patch does this by adding separate checks in each of the individual handlers. I filed https://issues.apache.org/jira/browse/KAFKA-12278 to try and handle this problem in a more general way, and so that we can also keep ApiVersions consistent. I would rather address this problem this problem in that JIRA if it is ok with you.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org