Skip to content

KAFKA-14402: Update AddPartitionsToTxn protocol to batch and handle verifyOnly requests #13231

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jolshan merged 18 commits into from
Mar 7, 2023
Merged

KAFKA-14402: Update AddPartitionsToTxn protocol to batch and handle verifyOnly requests #13231

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
18 commits
Select commit Hold shift + click to select a range
4f760d6
basic batching
jolshan Feb 3, 2023
70d11ba
cleaner batching
jolshan Feb 4, 2023
a8d6c00
bug fixes
jolshan Feb 6, 2023
47f0fff
add verifyOnly and auth changes
jolshan Feb 7, 2023
4c4d99e
more tests
jolshan Feb 10, 2023
4ad3df0
fixes
jolshan Feb 10, 2023
3f0c89c
update request json
jolshan Feb 21, 2023
791d56b
new builders
jolshan Feb 21, 2023
f8dbaa1
Fix up requests/responses
jolshan Feb 22, 2023
2c7d25c
Per Partition error codes for verify only
jolshan Feb 23, 2023
4aa0417
Merge branch 'trunk' of github.com:apache/kafka into kafka-14402
jolshan Feb 23, 2023
19dbfe6
unstable apis support
jolshan Feb 23, 2023
42486de
Fix tests
jolshan Feb 23, 2023
32ca9bb
just change the config for the test
jolshan Feb 24, 2023
a7abdae
Nits, changed to one callback, added kafka apis test for batched mode
jolshan Mar 1, 2023
fd1f92b
Redo callbacks by splitting into two methods, clean up error handling…
jolshan Mar 2, 2023
4589d36
test fix
jolshan Mar 3, 2023
ac16b13
Small changes
jolshan Mar 3, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
145 changes: 125 additions & 20 deletions clients/src/main/java/org/apache/kafka/common/requests/AddPartitionsToTxnRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -35,21 +44,43 @@ public class AddPartitionsToTxnRequest extends AbstractRequest {
private final AddPartitionsToTxnRequestData data;

private List<TopicPartition> cachedPartitions = null;

private Map<String, List<TopicPartition>> cachedPartitionsByTransaction = null;

private final short version;
Comment thread
jolshan marked this conversation as resolved.
Outdated

public static class Builder extends AbstractRequest.Builder<AddPartitionsToTxnRequest> {
public final AddPartitionsToTxnRequestData data;
public final boolean isClientRequest;

public Builder(final AddPartitionsToTxnRequestData data) {
// Only used for versions < 4
Comment thread
jolshan marked this conversation as resolved.
Outdated
Comment thread
jolshan marked this conversation as resolved.
Outdated
public Builder(String transactionalId,
long producerId,
short producerEpoch,
List<TopicPartition> partitions) {
super(ApiKeys.ADD_PARTITIONS_TO_TXN);
this.data = data;
this.isClientRequest = true;

AddPartitionsToTxnTopicCollection topics = compileTopics(partitions);

this.data = new AddPartitionsToTxnRequestData()
.setTransactionalId(transactionalId)
Comment thread
jolshan marked this conversation as resolved.
Outdated
.setProducerId(producerId)
.setProducerEpoch(producerEpoch)
.setTopics(topics);
Comment thread
jolshan marked this conversation as resolved.
Outdated
}

public Builder(final String transactionalId,
final long producerId,
final short producerEpoch,
final List<TopicPartition> partitions) {
public Builder(AddPartitionsToTxnTransactionCollection transactions,
boolean verifyOnly) {
super(ApiKeys.ADD_PARTITIONS_TO_TXN);
this.isClientRequest = false;

this.data = new AddPartitionsToTxnRequestData()
.setTransactions(transactions)
Comment thread
jolshan marked this conversation as resolved.
Outdated
.setVerifyOnly(verifyOnly);
}

private AddPartitionsToTxnTopicCollection compileTopics(final List<TopicPartition> partitions) {
Comment thread
jolshan marked this conversation as resolved.
Outdated
Map<String, List<Integer>> partitionMap = new HashMap<>();
for (TopicPartition topicPartition : partitions) {
String topicName = topicPartition.topic();
Expand All @@ -66,24 +97,22 @@ 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()));
.setName(partitionEntry.getKey())
.setPartitions(partitionEntry.getValue()));
Comment thread
jolshan marked this conversation as resolved.
Outdated
}

this.data = new AddPartitionsToTxnRequestData()
.setTransactionalId(transactionalId)
.setProducerId(producerId)
.setProducerEpoch(producerEpoch)
.setTopics(topics);
return topics;
}

@Override
public AddPartitionsToTxnRequest build(short version) {
return new AddPartitionsToTxnRequest(data, version);
short clampedVersion = (isClientRequest && version > 3) ? 3 : version;
Comment thread
jolshan marked this conversation as resolved.
Outdated
return new AddPartitionsToTxnRequest(data, clampedVersion);
}

// Only used for versions < 4
static List<TopicPartition> getPartitions(AddPartitionsToTxnRequestData data) {
List<TopicPartition> partitions = new ArrayList<>();

for (AddPartitionsToTxnTopic topicCollection : data.topics()) {
for (Integer partition : topicCollection.partitions()) {
partitions.add(new TopicPartition(topicCollection.name(), partition));
Expand All @@ -101,15 +130,61 @@ public String toString() {
public AddPartitionsToTxnRequest(final AddPartitionsToTxnRequestData data, short version) {
super(ApiKeys.ADD_PARTITIONS_TO_TXN, version);
this.data = data;
this.version = version;
}


// Only used for versions < 4
public List<TopicPartition> partitions() {
if (cachedPartitions != null) {
return cachedPartitions;
}
cachedPartitions = Builder.getPartitions(data);
return cachedPartitions;
}

private List<TopicPartition> partitionsForTransaction(String transaction) {
Comment thread
jolshan marked this conversation as resolved.
Outdated
Comment thread
jolshan marked this conversation as resolved.
Outdated
if (cachedPartitionsByTransaction == null) {
cachedPartitionsByTransaction = new HashMap<>();
}

return cachedPartitionsByTransaction.computeIfAbsent(transaction, txn -> {
List<TopicPartition> partitions = new ArrayList<>();
for (AddPartitionsToTxnTopic topicCollection : data.transactions().find(txn).topics()) {
for (Integer partition : topicCollection.partitions()) {
partitions.add(new TopicPartition(topicCollection.name(), partition));
}
}
return partitions;
});
}

public Map<String, List<TopicPartition>> partitionsByTransaction() {
if (cachedPartitionsByTransaction != null && cachedPartitionsByTransaction.size() == data.transactions().size()) {
Comment thread
jolshan marked this conversation as resolved.
Outdated
return cachedPartitionsByTransaction;
}

for (AddPartitionsToTxnTransaction transaction : data.transactions()) {
if (cachedPartitionsByTransaction == null || !cachedPartitionsByTransaction.containsKey(transaction.transactionalId())) {
partitionsForTransaction(transaction.transactionalId());
}
}
return cachedPartitionsByTransaction;
}

// 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.transactionalId())
.setProducerId(data.producerId())
.setProducerEpoch(data.producerEpoch())
.setTopics(data.topics()));
return singleTxn;
}

@Override
public AddPartitionsToTxnRequestData data() {
Expand All @@ -118,11 +193,41 @@ public AddPartitionsToTxnRequestData data() {

@Override
public AddPartitionsToTxnResponse getErrorResponse(int throttleTimeMs, Throwable e) {
final HashMap<TopicPartition, Errors> errors = new HashMap<>();
for (TopicPartition partition : partitions()) {
errors.put(partition, Errors.forException(e));
Errors error = Errors.forException(e);
if (version < 4) {
final HashMap<TopicPartition, Errors> errors = new HashMap<>();
for (TopicPartition partition : partitions()) {
errors.put(partition, error);
}
return new AddPartitionsToTxnResponse(throttleTimeMs, errors);
} else {
AddPartitionsToTxnResponseData response = new AddPartitionsToTxnResponseData();
Comment thread
jolshan marked this conversation as resolved.
Outdated
AddPartitionsToTxnResultCollection results = new AddPartitionsToTxnResultCollection();
for (AddPartitionsToTxnTransaction transaction : data().transactions()) {
results.add(errorResponseForTransaction(transaction.transactionalId(), error));
}
response.setResultsByTransaction(results);
response.setThrottleTimeMs(throttleTimeMs);
return new AddPartitionsToTxnResponse(response);
}
}

public AddPartitionsToTxnResult errorResponseForTransaction(String transactionalId, Errors e) {
AddPartitionsToTxnResult txnResult = new AddPartitionsToTxnResult().setTransactionalId(transactionalId);
AddPartitionsToTxnTopicResultCollection topicResults = new AddPartitionsToTxnTopicResultCollection();
for (AddPartitionsToTxnTopic topic : data.transactions().find(transactionalId).topics()) {
AddPartitionsToTxnTopicResult topicResult = new AddPartitionsToTxnTopicResult().setName(topic.name());
AddPartitionsToTxnPartitionResultCollection partitionResult = new AddPartitionsToTxnPartitionResultCollection();
for (Integer partition : topic.partitions()) {
partitionResult.add(new AddPartitionsToTxnPartitionResult()
.setPartitionIndex(partition)
.setErrorCode(e.code()));
}
topicResult.setResults(partitionResult);
topicResults.add(topicResult);
}
return new AddPartitionsToTxnResponse(throttleTimeMs, errors);
txnResult.setTopicResults(topicResults);
return txnResult;
}

public static AddPartitionsToTxnRequest parse(ByteBuffer buffer, short version) {
Expand Down
71 changes: 61 additions & 10 deletions clients/src/main/java/org/apache/kafka/common/requests/AddPartitionsToTxnResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.message.AddPartitionsToTxnResponseData;
import org.apache.kafka.common.message.AddPartitionsToTxnResponseData.AddPartitionsToTxnResult;
import org.apache.kafka.common.message.AddPartitionsToTxnResponseData.AddPartitionsToTxnPartitionResult;
import org.apache.kafka.common.message.AddPartitionsToTxnResponseData.AddPartitionsToTxnPartitionResultCollection;
import org.apache.kafka.common.message.AddPartitionsToTxnResponseData.AddPartitionsToTxnTopicResult;
Expand All @@ -27,7 +28,9 @@
import org.apache.kafka.common.protocol.Errors;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
Expand All @@ -49,28 +52,37 @@ public class AddPartitionsToTxnResponse extends AbstractResponse {
private final AddPartitionsToTxnResponseData data;

private Map<TopicPartition, Errors> cachedErrorsMap = null;

private Map<String, Map<TopicPartition, Errors>> cachedAllErrorsMap = null;

public AddPartitionsToTxnResponse(AddPartitionsToTxnResponseData data) {
super(ApiKeys.ADD_PARTITIONS_TO_TXN);
this.data = data;
}

// Only used for versions < 4
Comment thread
jolshan marked this conversation as resolved.
Outdated
public AddPartitionsToTxnResponse(int throttleTimeMs, Map<TopicPartition, Errors> errors) {
super(ApiKeys.ADD_PARTITIONS_TO_TXN);

this.data = new AddPartitionsToTxnResponseData()
.setThrottleTimeMs(throttleTimeMs)
.setResults(topicCollectionForErrors(errors));
}

private static AddPartitionsToTxnTopicResultCollection topicCollectionForErrors(Map<TopicPartition, Errors> errors) {
Map<String, AddPartitionsToTxnPartitionResultCollection> resultMap = new HashMap<>();

Comment thread
jolshan marked this conversation as resolved.
Outdated
for (Map.Entry<TopicPartition, Errors> entry : errors.entrySet()) {
TopicPartition topicPartition = entry.getKey();
String topicName = topicPartition.topic();

AddPartitionsToTxnPartitionResult partitionResult =
new AddPartitionsToTxnPartitionResult()
.setErrorCode(entry.getValue().code())
.setPartitionIndex(topicPartition.partition());
new AddPartitionsToTxnPartitionResult()
.setErrorCode(entry.getValue().code())
.setPartitionIndex(topicPartition.partition());

AddPartitionsToTxnPartitionResultCollection partitionResultCollection = resultMap.getOrDefault(
topicName, new AddPartitionsToTxnPartitionResultCollection()
topicName, new AddPartitionsToTxnPartitionResultCollection()
);

partitionResultCollection.add(partitionResult);
Expand All @@ -80,13 +92,14 @@ topicName, new AddPartitionsToTxnPartitionResultCollection()
AddPartitionsToTxnTopicResultCollection topicCollection = new AddPartitionsToTxnTopicResultCollection();
for (Map.Entry<String, AddPartitionsToTxnPartitionResultCollection> entry : resultMap.entrySet()) {
topicCollection.add(new AddPartitionsToTxnTopicResult()
.setName(entry.getKey())
.setResults(entry.getValue()));
.setName(entry.getKey())
.setResults(entry.getValue()));
}
return topicCollection;
}

this.data = new AddPartitionsToTxnResponseData()
.setThrottleTimeMs(throttleTimeMs)
.setResults(topicCollection);
public static AddPartitionsToTxnResult resultForTransaction(String transactionalId, Map<TopicPartition, Errors> errors) {
return new AddPartitionsToTxnResult().setTransactionalId(transactionalId).setTopicResults(topicCollectionForErrors(errors));
}

@Override
Expand All @@ -99,6 +112,7 @@ public void maybeSetThrottleTimeMs(int throttleTimeMs) {
data.setThrottleTimeMs(throttleTimeMs);
}

// Only used for versions < 4
Comment thread
jolshan marked this conversation as resolved.
Outdated
public Map<TopicPartition, Errors> errors() {
if (cachedErrorsMap != null) {
return cachedErrorsMap;
Expand All @@ -115,9 +129,46 @@ public Map<TopicPartition, Errors> errors() {
}
return cachedErrorsMap;
}

public Map<TopicPartition, Errors> errorsPerTransaction(String transactionalId) {
if (cachedAllErrorsMap == null) {
cachedAllErrorsMap = new HashMap<>();
}

return cachedAllErrorsMap.computeIfAbsent(transactionalId, txnId -> {
Map<TopicPartition, Errors> topicResults = new HashMap<>();
for (AddPartitionsToTxnTopicResult topicResult : data().resultsByTransaction().find(txnId).topicResults()) {
for (AddPartitionsToTxnPartitionResult partitionResult : topicResult.results()) {
topicResults.put(
new TopicPartition(topicResult.name(), partitionResult.partitionIndex()), Errors.forCode(partitionResult.errorCode()));
}
}
return topicResults;
});
}

public Map<String, Map<TopicPartition, Errors>> allErrors() {
if (cachedAllErrorsMap != null && cachedAllErrorsMap.size() == data.resultsByTransaction().size()) {
return cachedAllErrorsMap;
}

for (AddPartitionsToTxnResult result : this.data.resultsByTransaction()) {
if (cachedAllErrorsMap == null || !cachedAllErrorsMap.containsKey(result.transactionalId())) {
errorsPerTransaction(result.transactionalId());
}
}
return cachedAllErrorsMap;
}

@Override
public Map<Errors, Integer> errorCounts() {
if (data.resultsByTransaction().size() > 0) {
List<Errors> allErrors = new ArrayList<>();
allErrors().forEach((txnId, errors) ->
allErrors.addAll(errors.values())
);
return errorCounts(allErrors);
}
return errorCounts(errors().values());
}

Expand Down
30 changes: 24 additions & 6 deletions clients/src/main/resources/common/message/AddPartitionsToTxnRequest.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,17 +23,35 @@
// Version 2 adds the support for new error code PRODUCER_FENCED.
//
// Version 3 enables flexible versions.
"validVersions": "0-3",
//
// Version 4 adds VerifyOnly field to check if partitions are already in transaction and adds support to batch multiple transactions.
Comment thread
jolshan marked this conversation as resolved.
"validVersions": "0-4",
Comment thread
jolshan marked this conversation as resolved.
"flexibleVersions": "3+",
"fields": [
{ "name": "TransactionalId", "type": "string", "versions": "0+", "entityType": "transactionalId",
{ "name": "VerifyOnly", "type": "bool", "versions": "4+", "default": false,
Comment thread
jolshan marked this conversation as resolved.
Outdated
"about": "Boolean to signify if we want to check if the partition is in the transaction rather than add it." },
{ "name": "Transactions", "type": "[]AddPartitionsToTxnTransaction", "versions": "4+",
"about": "List of transactions to add partitions to.", "fields": [
{ "name": "TransactionalId", "type": "string", "versions": "4+", "mapKey": true, "entityType": "transactionalId",
"about": "The transactional id corresponding to the transaction."},
{ "name": "ProducerId", "type": "int64", "versions": "4+", "entityType": "producerId",
"about": "Current producer id in use by the transactional id." },
{ "name": "ProducerEpoch", "type": "int16", "versions": "4+",
"about": "Current epoch associated with the producer id." },
{ "name": "Topics", "type": "[]AddPartitionsToTxnTopic", "versions": "4+",
"about": "The partitions to add to the transaction." }
]},
{ "name": "TransactionalId", "type": "string", "versions": "0-3", "entityType": "transactionalId",
"about": "The transactional id corresponding to the transaction."},
{ "name": "ProducerId", "type": "int64", "versions": "0+", "entityType": "producerId",
{ "name": "ProducerId", "type": "int64", "versions": "0-3", "entityType": "producerId",
"about": "Current producer id in use by the transactional id." },
{ "name": "ProducerEpoch", "type": "int16", "versions": "0+",
{ "name": "ProducerEpoch", "type": "int16", "versions": "0-3",
"about": "Current epoch associated with the producer id." },
{ "name": "Topics", "type": "[]AddPartitionsToTxnTopic", "versions": "0+",
"about": "The partitions to add to the transaction.", "fields": [
{ "name": "Topics", "type": "[]AddPartitionsToTxnTopic", "versions": "0-3",
"about": "The partitions to add to the transaction." }
],
"commonStructs": [
{ "name": "AddPartitionsToTxnTopic", "versions": "0+", "fields": [
{ "name": "Name", "type": "string", "versions": "0+", "mapKey": true, "entityType": "topicName",
"about": "The name of the topic." },
{ "name": "Partitions", "type": "[]int32", "versions": "0+",
Expand Down
Loading