KAFKA-14402: Update AddPartitionsToTxn protocol to batch and handle verifyOnly requests #13231
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -19,7 +19,16 @@ | |
| import org.apache.kafka.common.TopicPartition; | ||
| import org.apache.kafka.common.message.AddPartitionsToTxnRequestData; | ||
| import org.apache.kafka.common.message.AddPartitionsToTxnRequestData.AddPartitionsToTxnTopic; | ||
| import org.apache.kafka.common.message.AddPartitionsToTxnRequestData.AddPartitionsToTxnTransaction; | ||
| import org.apache.kafka.common.message.AddPartitionsToTxnRequestData.AddPartitionsToTxnTransactionCollection; | ||
| import org.apache.kafka.common.message.AddPartitionsToTxnRequestData.AddPartitionsToTxnTopicCollection; | ||
| import org.apache.kafka.common.message.AddPartitionsToTxnResponseData; | ||
| import org.apache.kafka.common.message.AddPartitionsToTxnResponseData.AddPartitionsToTxnPartitionResult; | ||
| import org.apache.kafka.common.message.AddPartitionsToTxnResponseData.AddPartitionsToTxnPartitionResultCollection; | ||
| import org.apache.kafka.common.message.AddPartitionsToTxnResponseData.AddPartitionsToTxnResult; | ||
| import org.apache.kafka.common.message.AddPartitionsToTxnResponseData.AddPartitionsToTxnResultCollection; | ||
| import org.apache.kafka.common.message.AddPartitionsToTxnResponseData.AddPartitionsToTxnTopicResult; | ||
| import org.apache.kafka.common.message.AddPartitionsToTxnResponseData.AddPartitionsToTxnTopicResultCollection; | ||
| import org.apache.kafka.common.protocol.ApiKeys; | ||
| import org.apache.kafka.common.protocol.ByteBufferAccessor; | ||
| import org.apache.kafka.common.protocol.Errors; | ||
|
|
@@ -34,22 +43,40 @@ public class AddPartitionsToTxnRequest extends AbstractRequest { | |
|
|
||
| private final AddPartitionsToTxnRequestData data; | ||
|
|
||
| private final short version; | ||
|
jolshan marked this conversation as resolved.
Outdated
|
||
|
|
||
| public static class Builder extends AbstractRequest.Builder<AddPartitionsToTxnRequest> { | ||
| public final AddPartitionsToTxnRequestData data; | ||
|
|
||
| public static Builder forClient(String transactionalId, | ||
| long producerId, | ||
| short producerEpoch, | ||
| List<TopicPartition> partitions) { | ||
|
|
||
| AddPartitionsToTxnTopicCollection topics = buildTxnTopicCollection(partitions); | ||
|
|
||
| return new Builder(ApiKeys.ADD_PARTITIONS_TO_TXN.oldestVersion(), | ||
| (short) 3, | ||
| new AddPartitionsToTxnRequestData() | ||
| .setV3AndBelowTransactionalId(transactionalId) | ||
| .setV3AndBelowProducerId(producerId) | ||
| .setV3AndBelowProducerEpoch(producerEpoch) | ||
| .setV3AndBelowTopics(topics)); | ||
|
jolshan marked this conversation as resolved.
Outdated
|
||
| } | ||
|
|
||
| public static Builder forBroker(AddPartitionsToTxnTransactionCollection transactions) { | ||
| return new Builder((short) 4, ApiKeys.ADD_PARTITIONS_TO_TXN.latestVersion(), | ||
| new AddPartitionsToTxnRequestData() | ||
| .setTransactions(transactions)); | ||
| } | ||
|
|
||
| public Builder(short minVersion, short maxVersion, AddPartitionsToTxnRequestData data) { | ||
|
jolshan marked this conversation as resolved.
Outdated
|
||
| super(ApiKeys.ADD_PARTITIONS_TO_TXN, minVersion, maxVersion); | ||
|
|
||
| this.data = data; | ||
| } | ||
|
|
||
| private static AddPartitionsToTxnTopicCollection buildTxnTopicCollection(final List<TopicPartition> partitions) { | ||
| Map<String, List<Integer>> partitionMap = new HashMap<>(); | ||
| for (TopicPartition topicPartition : partitions) { | ||
| String topicName = topicPartition.topic(); | ||
|
|
@@ -66,32 +93,17 @@ public Builder(final String transactionalId, | |
| AddPartitionsToTxnTopicCollection topics = new AddPartitionsToTxnTopicCollection(); | ||
| for (Map.Entry<String, List<Integer>> partitionEntry : partitionMap.entrySet()) { | ||
| topics.add(new AddPartitionsToTxnTopic() | ||
| .setName(partitionEntry.getKey()) | ||
| .setPartitions(partitionEntry.getValue())); | ||
| } | ||
| return topics; | ||
| } | ||
|
|
||
| @Override | ||
| public AddPartitionsToTxnRequest build(short version) { | ||
| return new AddPartitionsToTxnRequest(data, version); | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return data.toString(); | ||
|
|
@@ -101,14 +113,7 @@ public String toString() { | |
| public AddPartitionsToTxnRequest(final AddPartitionsToTxnRequestData data, short version) { | ||
| super(ApiKeys.ADD_PARTITIONS_TO_TXN, version); | ||
| this.data = data; | ||
| this.version = version; | ||
| } | ||
|
|
||
| @Override | ||
|
|
@@ -118,11 +123,78 @@ public AddPartitionsToTxnRequestData data() { | |
|
|
||
| @Override | ||
| public AddPartitionsToTxnResponse getErrorResponse(int throttleTimeMs, Throwable e) { | ||
| Errors error = Errors.forException(e); | ||
| AddPartitionsToTxnResponseData response = new AddPartitionsToTxnResponseData(); | ||
| if (version < 4) { | ||
| response.setResultsByTopicV3AndBelow(errorResponseForTopics(data.v3AndBelowTopics(), error)); | ||
| } else { | ||
| AddPartitionsToTxnResultCollection results = new AddPartitionsToTxnResultCollection(); | ||
| for (AddPartitionsToTxnTransaction transaction : data().transactions()) { | ||
| results.add(errorResponseForTransaction(transaction.transactionalId(), error)); | ||
| } | ||
| response.setResultsByTransaction(results); | ||
| response.setErrorCode(error.code()); | ||
|
Member
There was a problem hiding this comment. When there is a global error, do we expect to set both the top level error and to return an error for each transaction? If we alway set both, what is the purpose of the top level error?
Member
Author
There was a problem hiding this comment. As discussed by @hachikuji and @guozhangwang the idea was if a top level error was set we could skip the rest of the handling. I just set all the fields to the same error for consistency. What is your suggestion? To remove these
Member
There was a problem hiding this comment. Yeah, I like the top level error code for that reason. However, we must be clear on how we want to use it. I suppose that we can only use it for cluster authorization failure and unexpected errors failing the entire request/response.
Member
Author
There was a problem hiding this comment. To clarify -- I'm not sure if being clear on how we want to use it (ie the errors it is used for) is the same question as if we want to include the error on the txns as well. As for the latter, I think we are more error prone if we set the error to none. I guess I'm just not sure what we are trying to accomplish by removing them from the lower fields. To be clear, I think I've also seen this pattern on other responses with top level errors.
Member
There was a problem hiding this comment. I was looking at the AlterPartition API as a reference. There, when we have a top level error, we don't set the partitions in the response at all. We basically save space. See here.
Member
Author
There was a problem hiding this comment. Ok.
Member
There was a problem hiding this comment. Don't forget to take the version into account here. We can only do this for v4.
Member
Author
There was a problem hiding this comment. Yup. This will only be done in v4 responses. So for now, we don't do any checks since the server side usage is upcoming in the next PR. As a side note, seems that AlterPartition includes the top level error in errorCounts so I did that here as well. |
||
| } | ||
| response.setThrottleTimeMs(throttleTimeMs); | ||
| return new AddPartitionsToTxnResponse(response); | ||
| } | ||
|
|
||
| public static List<TopicPartition> getPartitions(AddPartitionsToTxnTopicCollection topics) { | ||
| List<TopicPartition> partitions = new ArrayList<>(); | ||
|
|
||
| for (AddPartitionsToTxnTopic topicCollection : topics) { | ||
| for (Integer partition : topicCollection.partitions()) { | ||
| partitions.add(new TopicPartition(topicCollection.name(), partition)); | ||
| } | ||
| } | ||
| return partitions; | ||
| } | ||
|
|
||
| public Map<String, List<TopicPartition>> partitionsByTransaction() { | ||
| Map<String, List<TopicPartition>> partitionsByTransaction = new HashMap<>(); | ||
| for (AddPartitionsToTxnTransaction transaction : data.transactions()) { | ||
| List<TopicPartition> partitions = getPartitions(transaction.topics()); | ||
| partitionsByTransaction.put(transaction.transactionalId(), partitions); | ||
| } | ||
| return partitionsByTransaction; | ||
| } | ||
|
|
||
| // Takes a version 3 or below request and returns a v4+ singleton (one transaction ID) request. | ||
| public AddPartitionsToTxnRequest normalizeRequest() { | ||
| return new AddPartitionsToTxnRequest(new AddPartitionsToTxnRequestData().setTransactions(singletonTransaction()), version); | ||
| } | ||
|
|
||
| private AddPartitionsToTxnTransactionCollection singletonTransaction() { | ||
| AddPartitionsToTxnTransactionCollection singleTxn = new AddPartitionsToTxnTransactionCollection(); | ||
| singleTxn.add(new AddPartitionsToTxnTransaction() | ||
| .setTransactionalId(data.v3AndBelowTransactionalId()) | ||
| .setProducerId(data.v3AndBelowProducerId()) | ||
| .setProducerEpoch(data.v3AndBelowProducerEpoch()) | ||
| .setTopics(data.v3AndBelowTopics())); | ||
| return singleTxn; | ||
| } | ||
|
|
||
| public AddPartitionsToTxnResult errorResponseForTransaction(String transactionalId, Errors e) { | ||
| AddPartitionsToTxnResult txnResult = new AddPartitionsToTxnResult().setTransactionalId(transactionalId); | ||
| AddPartitionsToTxnTopicResultCollection topicResults = errorResponseForTopics(data.transactions().find(transactionalId).topics(), e); | ||
| txnResult.setTopicResults(topicResults); | ||
| return txnResult; | ||
| } | ||
|
|
||
| private AddPartitionsToTxnTopicResultCollection errorResponseForTopics(AddPartitionsToTxnTopicCollection topics, Errors e) { | ||
| AddPartitionsToTxnTopicResultCollection topicResults = new AddPartitionsToTxnTopicResultCollection(); | ||
| for (AddPartitionsToTxnTopic topic : topics) { | ||
| AddPartitionsToTxnTopicResult topicResult = new AddPartitionsToTxnTopicResult().setName(topic.name()); | ||
| AddPartitionsToTxnPartitionResultCollection partitionResult = new AddPartitionsToTxnPartitionResultCollection(); | ||
| for (Integer partition : topic.partitions()) { | ||
| partitionResult.add(new AddPartitionsToTxnPartitionResult() | ||
| .setPartitionIndex(partition) | ||
| .setPartitionErrorCode(e.code())); | ||
| } | ||
| topicResult.setResultsByPartition(partitionResult); | ||
| topicResults.add(topicResult); | ||
| } | ||
| return topicResults; | ||
| } | ||
|
|
||
| public static AddPartitionsToTxnRequest parse(ByteBuffer buffer, short version) { | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
nit: I suppose that
errorsshould never benullhere. I wonder if we should still check it. What do you think?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
What should we do if the check fails? Just have a better error message thrown?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yeah, the check is probably not necessary. By the way, I find the idea of having
V3_AND_BELOW_TXN_IDfor old version a bit confusing. I was wondering if usingaddPartitionsToTxnResponse.data().resultsByTopicV3AndBelow()would be a better alternative here. We only iterate over the Map so the Map is not strictly required here. Have you considered something like this?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I was told not to have v3 and below specific methods from Jason because the v3 case should generalize to a single version of the v4 case and that should make it easy to use methods for both.
However, if we really think this is an issue. I guess we can change the approach again. I'm just not sure the experience of errors only applying to v4+. Any ideas there besides changing the method name to express it should only be used in v4+?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Understood. Let's keep it as it is then.
I agree that v3 case should generalized to a single item of the v4 case. It is just unfortunate that we don't have the transaction id in v3 response so we have to use an empty string for it. I suppose that it is the way it is.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yeah. It really is unfortunate. 😞