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 all 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
4 changes: 2 additions & 2 deletions clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1052,7 +1052,7 @@ private TxnRequestHandler addPartitionsToTransactionHandler() {
pendingPartitionsInTransaction.addAll(newPartitionsInTransaction);
newPartitionsInTransaction.clear();
AddPartitionsToTxnRequest.Builder builder =
new AddPartitionsToTxnRequest.Builder(transactionalId,
AddPartitionsToTxnRequest.Builder.forClient(transactionalId,
producerIdAndEpoch.producerId,
producerIdAndEpoch.epoch,
new ArrayList<>(pendingPartitionsInTransaction));
Expand Down Expand Up @@ -1328,7 +1328,7 @@ Priority priority() {
@Override
public void handleResponse(AbstractResponse response) {
AddPartitionsToTxnResponse addPartitionsToTxnResponse = (AddPartitionsToTxnResponse) response;
Map<TopicPartition, Errors> errors = addPartitionsToTxnResponse.errors();
Map<TopicPartition, Errors> errors = addPartitionsToTxnResponse.errors().get(AddPartitionsToTxnResponse.V3_AND_BELOW_TXN_ID);

@dajac dajac Feb 27, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: I suppose that errors should never be null here. I wonder if we should still check it. What do you think?

@jolshan jolshan Mar 1, 2023

Copy link
Copy Markdown
Member Author

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?

@dajac dajac Mar 2, 2023

Copy link
Copy Markdown
Member

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_ID for old version a bit confusing. I was wondering if using addPartitionsToTxnResponse.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?

@jolshan jolshan Mar 2, 2023

Copy link
Copy Markdown
Member Author

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+?

@dajac dajac Mar 2, 2023

Copy link
Copy Markdown
Member

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.

@jolshan jolshan Mar 2, 2023

Copy link
Copy Markdown
Member Author

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. 😞

boolean hasPartitionErrors = false;
Set<String> unauthorizedTopics = new HashSet<>();
retryBackoffMs = TransactionManager.this.retryBackoffMs;
Expand Down
142 changes: 102 additions & 40 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,15 @@
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.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 @@ -34,22 +42,37 @@ public class AddPartitionsToTxnRequest extends AbstractRequest {

private final AddPartitionsToTxnRequestData data;

private List<TopicPartition> cachedPartitions = null;

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));
}

public static Builder forBroker(AddPartitionsToTxnTransactionCollection transactions) {
return new Builder((short) 4, ApiKeys.ADD_PARTITIONS_TO_TXN.latestVersion(),
new AddPartitionsToTxnRequestData()
.setTransactions(transactions));
}

private Builder(short minVersion, short maxVersion, AddPartitionsToTxnRequestData data) {
super(ApiKeys.ADD_PARTITIONS_TO_TXN, minVersion, maxVersion);

public Builder(final AddPartitionsToTxnRequestData data) {
super(ApiKeys.ADD_PARTITIONS_TO_TXN);
this.data = data;
}

public Builder(final String transactionalId,
final long producerId,
final short producerEpoch,
final List<TopicPartition> partitions) {
super(ApiKeys.ADD_PARTITIONS_TO_TXN);

private static AddPartitionsToTxnTopicCollection buildTxnTopicCollection(final List<TopicPartition> partitions) {
Map<String, List<Integer>> partitionMap = new HashMap<>();
for (TopicPartition topicPartition : partitions) {
String topicName = topicPartition.topic();
Expand All @@ -66,32 +89,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()));
.setName(partitionEntry.getKey())
.setPartitions(partitionEntry.getValue()));
}

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);
}

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));
}
}
return partitions;
}

@Override
public String toString() {
return data.toString();
Expand All @@ -103,26 +111,80 @@ public AddPartitionsToTxnRequest(final AddPartitionsToTxnRequestData data, short
this.data = data;
}

public List<TopicPartition> partitions() {
if (cachedPartitions != null) {
return cachedPartitions;
}
cachedPartitions = Builder.getPartitions(data);
return cachedPartitions;
}

@Override
public AddPartitionsToTxnRequestData data() {
return 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);
AddPartitionsToTxnResponseData response = new AddPartitionsToTxnResponseData();
if (version() < 4) {
response.setResultsByTopicV3AndBelow(errorResponseForTopics(data.v3AndBelowTopics(), error));
} else {
response.setErrorCode(error.code());
}
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 new AddPartitionsToTxnResponse(throttleTimeMs, errors);
return topicResults;
}

public static AddPartitionsToTxnRequest parse(ByteBuffer buffer, short version) {
Expand Down
90 changes: 57 additions & 33 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 @@ -48,29 +51,51 @@ public class AddPartitionsToTxnResponse extends AbstractResponse {

private final AddPartitionsToTxnResponseData data;

private Map<TopicPartition, Errors> cachedErrorsMap = null;
public static final String V3_AND_BELOW_TXN_ID = "";

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

public AddPartitionsToTxnResponse(int throttleTimeMs, Map<TopicPartition, Errors> errors) {
super(ApiKeys.ADD_PARTITIONS_TO_TXN);
@Override
public int throttleTimeMs() {
return data.throttleTimeMs();
}

@Override
public void maybeSetThrottleTimeMs(int throttleTimeMs) {
data.setThrottleTimeMs(throttleTimeMs);
}

public Map<String, Map<TopicPartition, Errors>> errors() {
Map<String, Map<TopicPartition, Errors>> errorsMap = new HashMap<>();

if (!this.data.resultsByTopicV3AndBelow().isEmpty()) {
errorsMap.put(V3_AND_BELOW_TXN_ID, errorsForTransaction(this.data.resultsByTopicV3AndBelow()));
}

for (AddPartitionsToTxnResult result : this.data.resultsByTransaction()) {
errorsMap.put(result.transactionalId(), errorsForTransaction(result.topicResults()));
}

return errorsMap;
}

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

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()
.setPartitionErrorCode(entry.getValue().code())
.setPartitionIndex(topicPartition.partition());

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

partitionResultCollection.add(partitionResult);
Expand All @@ -80,45 +105,44 @@ 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())
.setResultsByPartition(entry.getValue()));
}

this.data = new AddPartitionsToTxnResponseData()
.setThrottleTimeMs(throttleTimeMs)
.setResults(topicCollection);
return topicCollection;
}

@Override
public int throttleTimeMs() {
return data.throttleTimeMs();
public static AddPartitionsToTxnResult resultForTransaction(String transactionalId, Map<TopicPartition, Errors> errors) {
return new AddPartitionsToTxnResult().setTransactionalId(transactionalId).setTopicResults(topicCollectionForErrors(errors));
}

@Override
public void maybeSetThrottleTimeMs(int throttleTimeMs) {
data.setThrottleTimeMs(throttleTimeMs);
public AddPartitionsToTxnTopicResultCollection getTransactionTopicResults(String transactionalId) {
return data.resultsByTransaction().find(transactionalId).topicResults();
}

public Map<TopicPartition, Errors> errors() {
if (cachedErrorsMap != null) {
return cachedErrorsMap;
}

cachedErrorsMap = new HashMap<>();

for (AddPartitionsToTxnTopicResult topicResult : this.data.results()) {
for (AddPartitionsToTxnPartitionResult partitionResult : topicResult.results()) {
cachedErrorsMap.put(new TopicPartition(
topicResult.name(), partitionResult.partitionIndex()),
Errors.forCode(partitionResult.errorCode()));
public static Map<TopicPartition, Errors> errorsForTransaction(AddPartitionsToTxnTopicResultCollection topicCollection) {
Map<TopicPartition, Errors> topicResults = new HashMap<>();
for (AddPartitionsToTxnTopicResult topicResult : topicCollection) {
for (AddPartitionsToTxnPartitionResult partitionResult : topicResult.resultsByPartition()) {
topicResults.put(
new TopicPartition(topicResult.name(), partitionResult.partitionIndex()), Errors.forCode(partitionResult.partitionErrorCode()));
}
}
return cachedErrorsMap;
return topicResults;
}

@Override
public Map<Errors, Integer> errorCounts() {
return errorCounts(errors().values());
List<Errors> allErrors = new ArrayList<>();

// If we are not using this field, we have request 4 or later
if (this.data.resultsByTopicV3AndBelow().isEmpty()) {
allErrors.add(Errors.forCode(data.errorCode()));

@dajac dajac Mar 3, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: Should we use updateErrorCounts from AbstractResponse instead of creating allErrors?

@jolshan jolshan Mar 3, 2023

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I create allErrors because I use addAll for the individual transactions. I can place this code after I create error counts, but it doesn't really seem to accomplish much.

}

errors().forEach((txnId, errors) ->
allErrors.addAll(errors.values())
);
return errorCounts(allErrors);
}

@Override
Expand Down
Loading