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 2020/12/30 22:44:43 UTC
[GitHub] [kafka] hachikuji opened a new pull request #9802: KAFKA-10894; Ensure PartitionInfo replicas are not null in client quota callback
hachikuji opened a new pull request #9802:
URL: https://github.com/apache/kafka/pull/9802
Previously offline replicas were included as `null` in the array of replicas in `PartitionInfo` when populated by the `MetadataCache` for the purpose of the client quota callback. This patch instead initializes empty non-null nodes, which is consistent with how `PartitionInfo` is constructed by the clients in `MetadataResponse`.
### 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] hachikuji commented on pull request #9802: KAFKA-10894; Ensure PartitionInfo replicas are not null in client quota callback
Posted by GitBox <gi...@apache.org>.
hachikuji commented on pull request #9802:
URL: https://github.com/apache/kafka/pull/9802#issuecomment-754302596
@chia7712 Good question about `leader()`. I looked into usage and it seems we do not have consistent `isEmpty` checks, so I'm a little concerned about compatibility if we change that. To be honest, I do not like the use of the empty sentinel. It seems like a dangerous practice and I'm not too eager to expand its usage. However, given the choice between returning a list of replica nodes which contains null values (what is done today for this callback) and making use of the empty sentinel in the same way that it is already used in the client logic, the latter seems preferable. Does that make sense?
----------------------------------------------------------------
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 #9802: KAFKA-10894; Ensure PartitionInfo replicas are not null in client quota callback
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9802:
URL: https://github.com/apache/kafka/pull/9802#discussion_r552089415
##########
File path: core/src/main/scala/kafka/server/MetadataCache.scala
##########
@@ -266,8 +266,16 @@ class MetadataCache(brokerId: Int) extends Logging {
def getClusterMetadata(clusterId: String, listenerName: ListenerName): Cluster = {
val snapshot = metadataSnapshot
- val nodes = snapshot.aliveNodes.map { case (id, nodes) => (id, nodes.get(listenerName).orNull) }
- def node(id: Integer): Node = nodes.get(id.toLong).orNull
+ val nodes = snapshot.aliveNodes.flatMap { case (id, nodesByListener) =>
+ nodesByListener.get(listenerName).map { node =>
+ id -> node
+ }
+ }
+
+ def node(id: Integer): Node = {
+ nodes.getOrElse(id.toLong, new Node(id, "", -1))
Review comment:
I think what we are implementing here is what is documented in `PartitionInfo`:
```java
/**
* The complete set of replicas for this partition regardless of whether they are alive or up-to-date
*/
public Node[] replicas() {
return replicas;
}
```
Even the internal logic in `PartitionInfo` assumes non-null values in the array (I caught this issue because of an NPE in the `toString`). We probably can't 100% guarantee that all quota callbacks have appropriate `isEmpty` checks, but my guess is that usage of this plugin is rare in any case and reliance on the replica set is less likely than reliance on the leader. All in all, I think we're better fixing the bug.
----------------------------------------------------------------
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 #9802: KAFKA-10894; Ensure PartitionInfo replicas are not null in client quota callback
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9802:
URL: https://github.com/apache/kafka/pull/9802#discussion_r551641235
##########
File path: core/src/main/scala/kafka/server/MetadataCache.scala
##########
@@ -266,8 +266,16 @@ class MetadataCache(brokerId: Int) extends Logging {
def getClusterMetadata(clusterId: String, listenerName: ListenerName): Cluster = {
val snapshot = metadataSnapshot
- val nodes = snapshot.aliveNodes.map { case (id, nodes) => (id, nodes.get(listenerName).orNull) }
- def node(id: Integer): Node = nodes.get(id.toLong).orNull
+ val nodes = snapshot.aliveNodes.flatMap { case (id, nodesByListener) =>
+ nodesByListener.get(listenerName).map { node =>
+ id -> node
+ }
+ }
+
+ def node(id: Integer): Node = {
+ nodes.getOrElse(id.toLong, new Node(id, "", -1))
Review comment:
That's not a bad idea. I think `Node` is public, though we haven't always been strict about requiring KIPs for new methods. A couple options are to add a new constructor or a static factory. We could also add a utility somewhere, but that feels a little clumsy. Thoughts?
----------------------------------------------------------------
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] chia7712 commented on a change in pull request #9802: KAFKA-10894; Ensure PartitionInfo replicas are not null in client quota callback
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9802:
URL: https://github.com/apache/kafka/pull/9802#discussion_r551688940
##########
File path: core/src/main/scala/kafka/server/MetadataCache.scala
##########
@@ -266,8 +266,16 @@ class MetadataCache(brokerId: Int) extends Logging {
def getClusterMetadata(clusterId: String, listenerName: ListenerName): Cluster = {
val snapshot = metadataSnapshot
- val nodes = snapshot.aliveNodes.map { case (id, nodes) => (id, nodes.get(listenerName).orNull) }
- def node(id: Integer): Node = nodes.get(id.toLong).orNull
+ val nodes = snapshot.aliveNodes.flatMap { case (id, nodesByListener) =>
+ nodesByListener.get(listenerName).map { node =>
+ id -> node
+ }
+ }
+
+ def node(id: Integer): Node = {
+ nodes.getOrElse(id.toLong, new Node(id, "", -1))
Review comment:
> I think Node is public, though we haven't always been strict about requiring KIPs for new methods. A couple options are to add a new constructor or a static factory. We could also add a utility somewhere, but that feels a little clumsy. Thoughts?
You are right.
For another, is this a kind of behavior change to ```ClientQuotaCallback``` (if custom ```ClientQuotaCallback``` does depend on the ```null``` value)? If so, how we keep the compatibility or should we offer a helper method to enable ```ClientQuotaCallback``` to validate the "new Node(id, "", -1)"?
----------------------------------------------------------------
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] chia7712 commented on a change in pull request #9802: KAFKA-10894; Ensure PartitionInfo replicas are not null in client quota callback
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9802:
URL: https://github.com/apache/kafka/pull/9802#discussion_r550409107
##########
File path: core/src/main/scala/kafka/server/MetadataCache.scala
##########
@@ -266,8 +266,16 @@ class MetadataCache(brokerId: Int) extends Logging {
def getClusterMetadata(clusterId: String, listenerName: ListenerName): Cluster = {
val snapshot = metadataSnapshot
- val nodes = snapshot.aliveNodes.map { case (id, nodes) => (id, nodes.get(listenerName).orNull) }
- def node(id: Integer): Node = nodes.get(id.toLong).orNull
+ val nodes = snapshot.aliveNodes.flatMap { case (id, nodesByListener) =>
+ nodesByListener.get(listenerName).map { node =>
+ id -> node
+ }
+ }
+
+ def node(id: Integer): Node = {
+ nodes.getOrElse(id.toLong, new Node(id, "", -1))
Review comment:
How about using a helper method to create ```new Node(id, "", -1)``` ? It can be used by ```MetadataResponse``` also.
----------------------------------------------------------------
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