KAFKA-2945: CreateTopic - protocol and server side implementation#626
KAFKA-2945: CreateTopic - protocol and server side implementation#626granthenke wants to merge 2 commits into
Conversation
There was a problem hiding this comment.
A helpful convention we've adopted for some of the other apis is to list all of the possible error codes in comments in the response object.
There was a problem hiding this comment.
This seams useful, but also very prone to become outdated and invalid since the response is by no means tied to the implementation. And an Unknown error is always a possibility. If we want to make this guarantee we should error during serialization or at least log a warning or something.
There was a problem hiding this comment.
Totally agree on the potential to become outdated. I think I've already seen this for some APIs. It would be nice to have a more formal way of handling it, but that seems a little out of scope here, so documenting the error codes would be helpful until we have that.
There was a problem hiding this comment.
Happy to add the comments
17b7905 to
5dce806
Compare
|
@granthenke Overall LGTM. Are you planning to add integration/system tests here or in another ticket? Also, do you have a plan for an AdminClient? I created one in kafka.admin as part of KIP-40, but it was really a stopgap to get 0.9 out. I'm hoping we can have implement a proper client in kafka-clients. I'm happy to help with any of this since I also want to push KIP-4 through. |
|
@hachikuji Yes, I need to update the KIP-4 wiki and jira a bit more yet. But the goal is more or less 4 phased:
I will be sure to reach out for any help needed. Review on KIP-4 updates/design (when I update) and code review is always a great help. Thanks for the review! |
|
I have a silly question : create topic is implemented by Admin at 0.8.X ? It's looks like redesgin ? |
|
@darionyaphet You can read about KIP-4 here. I need to update that page, but the general information is there. This is implementing the ability to create topics via Kafka instead of going directly to Zookeeper. Functionality that does not exist today. |
There was a problem hiding this comment.
Do we need to validate that we don't have two conflicting requests for same topic?
There was a problem hiding this comment.
Since the list of topics stored in a Map in the Request object we are guaranteed to get only one instance of a given topic.
|
Small comment regarding conflicts and you'll need to rebase. |
There was a problem hiding this comment.
The fact that we need to add a comment to describe a single method call implies that we are doing something wrong. Does the AdminUtils method name need to be so obscure?
There was a problem hiding this comment.
tough call...it's technically accurate.
5dce806 to
0dead6f
Compare
|
Rebased, added some small fixes for the comments, and squashed the commits. |
There was a problem hiding this comment.
nit: Group o.a.k.* imports and kafka.* imports.
There was a problem hiding this comment.
I only added one import line here. It is next to another o.a.k import and sorted alphabetically. Should I be refactoring the all imports in any touched file?
|
LGTM overall. One general comment (may not need to be included in this PR) is that for some cases the client may want to receive the response not immediately after it is written to ZK, but until it has been successfully created (i.e. https://issues.apache.org/jira/browse/KAFKA-1125). Hence we may want to add another boolean field in the AbstractAdminRequest (where other requests like CreateTopicRequest extends from) indicating whether we want to be blocked or not. |
|
Thanks @granthenke for your reply. Yeah I think setting a timeout value in the |
9765397 to
15ff506
Compare
|
@guozhangwang @gwenshap @ijuma @hachikuji |
|
@granthenke : Thanks for the patch. It seems that you added the logic in KafkaApis to block a createTopicRequest until the metadata of the topic is propagated to the broker to which the createTopicRequest is sent. However, from a client's perspective, if it really wants to make sure that the topic is available, it has to wait for the metadata to be propagated to every broker since the metadata request can be issued to any broker. There are a couple ways to do that: (1) forward the createTopicRequest to the controller and let the controller block until it propagates the metadata to every broker; (2) if this is just for admin purpose, we can implement the logic in the admin client to check the metadata on every broker after topic creation. (1) is a bit involved since we probably have to clean up the controller a bit to facilitate that. If we want to do (2) for now, there is no need to block in the createTopicRequest. Alternatively, we can just make createTopicRequest async for now (i.e., the topic's metadata is not guaranteed to be propagated to every broker after the request is responded.). |
|
@junrao Thanks for taking a look. Good point, we really need to be sure its propagated everywhere. I need to look at what it would take to implement, but I would really like to do it right this time. Especially given that a bunch of other admin requests may need to follow a similar pattern. For example I have some WIP code for for KAFKA-2946: DeleteTopic - protocol and server side implementation that leverages the same code: You mentioned forwarding the requests to the controller. Another option is to force admin requests like this to go to the active controller and block until it propagates the metadata. This would require the MetadataResponse to the tell the client which broker is the controller. Is there a preference in which approach is taken? |
|
@granthenke : For requests like produce/fetch, it makes sense to send the requests directly to the leader broker, instead of forwarding since there are typically lots of those requests. For admin requests like create/delete, forwarding is less of an issue since there are a lot fewer of those requests. Being able to talk to any broker also has the benefit that it simplifies the client side logic. Our client code uses non-blocking socket. Requiring finding the controller broker first means that we have to track an additional state in the client, which makes the client implementation more complicated. So, overall, forwarding admin requests to the controller is probably better. We have to think through the following things: (1) we probably need a separate purgatory while waiting on the response if forwarding is needed. (2) Currently, the controller code is a bit messy and we plan to clean it up in the future (e.g., KAFKA-3210). So, ideally, whatever changes we make in the controller shouldn't be too intrusive or making the logic much more complicated than it is now. (3) If we make create/delete block, we probably should also make alter block as well. This is even tricker. First, currently, most topic configs are stored in ZK and propagated through ZK watchers to the broker directly w/o going through the controller. So, not sure how we make sure that the topic config change is propagated to all brokers unless we talk to every broker directly. Alternatively, we can change config propagation so that it also goes through the controller. However, this requires inter broker protocol changes. Second, if one changes the replica assignment, we have to copy data across brokers. This can take arbitrarily long. So, it probably doesn't make sense to have a blocking request in this case. The client will have to keep checking periodically to see if this is done. Given all the above, that's why initially we thought it's easier to start with just the async version of the those requests. If we want to go directly to the blocking implementation. It will be useful to think through the plan for each type of admin request. |
|
@junrao I agree that the various scenarios for blocking are complex. But I also agree with @guozhangwang that doing it on the broker side is valuable. We will need to solve all the same problems in the AdminClient, and then so will every other client implementation. Also, changing to a blocking implementation later will likely require wire protocol changes. I worry about the client, or brokers, polling every broker to validate the metadata is updated. For large clusters that may not scale and the implementation could be complex. For that reason, having the controller be the definitive authority on the state of the cluster would be useful. Since the MetadataRequest is a sort of "DescribeTopic" topic request. Would it make sense for that request to be routed to the controller as well? I agree it will be useful to think through the details of each admin request and how the forwarding logic would work as well. I will try some WIP implementations and update the KIP Wiki with the most up to date information and proposals so we can discuss it in the mailing list and on the KIP calls further. |
|
@granthenke : I am not sure about forwarding MetadataRequest to the controller. MetadataRequest is more frequent since there is periodic metadata refresh in the clients. That's why we maintain a metadata cache per broker. I agree that blocking on the broker side is useful for admin requests that are quick. I just want to point out this wasn't scoped during the original KIP-4 proposal. It's fine if we want to pick this up now. We can have a followup KIP discussion once we have a proposal on the implementation. |
|
@junrao I am happy to stick with what was discussed and go with the async version. The change to blocking was based on feedback on my pull requests. I am also happy to create a follow up KIP to implement a blocking version once this work is done. To reduce migration/compatibility pain, is it worth maintaining the timeout field in the wire protocol, and only block until the local broker is updated for now? (and document that) Regardless the KIP-4 wiki needs to be updated. So I will work on that. |
|
@junrao Should I leave this patch as is and review based on a local blocking timeout with plans to fully block in the future? Or should I revert to completely async? I don't feel strongly either way. My priority is to get this in, so I can create the other needed protocol messages and the admin client asap. |
|
@granthenke : My preference is that if we want to do blocking, we do this consistently across all admin apis and do that in a state that we want in the long term. The local blocking implementation is probably not we want in the long term and doesn't have the full blocking semantic. Given that, for this patch, I'd recommend that we just go with the async implementation for now. |
|
@junrao Perfect. I will submit an updated patch today to change back to async. I have created KAFKA-3249 to track future discussions about a blocking option. |
1bd3c8c to
9801638
Compare
|
@guozhangwang @gwenshap @ijuma @hachikuji @junrao |
d7db4a8 to
57d26e5
Compare
|
Rebased on trunk |
|
For anyone interested I have also updated my WIP delete branch that depends on this PR: granthenke/kafka@create-wire...granthenke:delete-wire |
|
@guozhangwang @gwenshap @ijuma @hachikuji @junrao |
| INVALID_REPLICA_ASSIGNMENT(36, | ||
| new InvalidReplicaAssignmentException("Replica assignment is invalid.")), | ||
| INVALID_ENTITY_CONFIG(37, | ||
| new InvalidEntityConfigurationException("Entity configuration is invalid.")); |
There was a problem hiding this comment.
Could we add those to ErrorMapping.scala as well? If they are not used in scala, just add them as place holders in the comment to reserve the error number.
Also, would it be simpler to rename InvalidEntityConfigurationException to just InvalidConfigurationException?
There was a problem hiding this comment.
I will add the comments. I didn't add it because all server code has been migrated to use Errors.java in KAFKA-2929: Migrate duplicate error mapping functionality, and the existing scala clients don't need these new codes.
|
@junrao Thanks for the review! I have updated the patch to address your comments. |
MINOR: upgrade netty version to 4.1.68.final
7 participants
No description provided.