Skip to content

KAFKA-8618: Replace Txn marker with automated protocol #7039

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
mimaison merged 2 commits into from
Mar 19, 2020
Merged

KAFKA-8618: Replace Txn marker with automated protocol #7039

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
9fefdcf
add txn marker auto
Jul 5, 2019
7d087fd
Remove extra empty line
mimaison Mar 19, 2020
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
8 changes: 4 additions & 4 deletions clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,6 +95,8 @@
import org.apache.kafka.common.message.UpdateMetadataResponseData;
import org.apache.kafka.common.message.TxnOffsetCommitRequestData;
import org.apache.kafka.common.message.TxnOffsetCommitResponseData;
import org.apache.kafka.common.message.WriteTxnMarkersRequestData;
import org.apache.kafka.common.message.WriteTxnMarkersResponseData;
import org.apache.kafka.common.protocol.types.Schema;
import org.apache.kafka.common.protocol.types.SchemaException;
import org.apache.kafka.common.protocol.types.Struct;
Expand All @@ -120,8 +122,6 @@
import org.apache.kafka.common.requests.OffsetsForLeaderEpochResponse;
import org.apache.kafka.common.requests.ProduceRequest;
import org.apache.kafka.common.requests.ProduceResponse;
import org.apache.kafka.common.requests.WriteTxnMarkersRequest;
import org.apache.kafka.common.requests.WriteTxnMarkersResponse;

import java.nio.ByteBuffer;
import java.util.concurrent.atomic.AtomicBoolean;
Expand Down Expand Up @@ -177,8 +177,8 @@ public Struct parseResponse(short version, ByteBuffer buffer) {
ADD_OFFSETS_TO_TXN(25, "AddOffsetsToTxn", false, RecordBatch.MAGIC_VALUE_V2, AddOffsetsToTxnRequest.schemaVersions(),
AddOffsetsToTxnResponse.schemaVersions()),
END_TXN(26, "EndTxn", false, RecordBatch.MAGIC_VALUE_V2, EndTxnRequestData.SCHEMAS, EndTxnResponseData.SCHEMAS),
WRITE_TXN_MARKERS(27, "WriteTxnMarkers", true, RecordBatch.MAGIC_VALUE_V2, WriteTxnMarkersRequest.schemaVersions(),
WriteTxnMarkersResponse.schemaVersions()),
WRITE_TXN_MARKERS(27, "WriteTxnMarkers", true, RecordBatch.MAGIC_VALUE_V2, WriteTxnMarkersRequestData.SCHEMAS,
WriteTxnMarkersResponseData.SCHEMAS),
TXN_OFFSET_COMMIT(28, "TxnOffsetCommit", false, RecordBatch.MAGIC_VALUE_V2, TxnOffsetCommitRequestData.SCHEMAS,
TxnOffsetCommitResponseData.SCHEMAS),
DESCRIBE_ACLS(29, "DescribeAcls", DescribeAclsRequestData.SCHEMAS, DescribeAclsResponseData.SCHEMAS),
Expand Down
2 changes: 1 addition & 1 deletion clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -135,7 +135,7 @@ public static AbstractResponse parseResponse(ApiKeys apiKey, Struct struct, shor
case END_TXN:
return new EndTxnResponse(struct, version);
case WRITE_TXN_MARKERS:
return new WriteTxnMarkersResponse(struct);
return new WriteTxnMarkersResponse(struct, version);
case TXN_OFFSET_COMMIT:
return new TxnOffsetCommitResponse(struct, version);
case DESCRIBE_ACLS:
Expand Down
192 changes: 72 additions & 120 deletions clients/src/main/java/org/apache/kafka/common/requests/WriteTxnMarkersRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,13 +17,12 @@
package org.apache.kafka.common.requests;

import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.message.WriteTxnMarkersRequestData;
import org.apache.kafka.common.message.WriteTxnMarkersRequestData.WritableTxnMarker;
import org.apache.kafka.common.message.WriteTxnMarkersRequestData.WritableTxnMarkerTopic;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.protocol.types.ArrayOf;
import org.apache.kafka.common.protocol.types.Field;
import org.apache.kafka.common.protocol.types.Schema;
import org.apache.kafka.common.protocol.types.Struct;
import org.apache.kafka.common.utils.CollectionUtils;

import java.nio.ByteBuffer;
import java.util.ArrayList;
Expand All @@ -32,40 +31,7 @@
import java.util.Map;
import java.util.Objects;

import static org.apache.kafka.common.protocol.CommonFields.TOPIC_NAME;
import static org.apache.kafka.common.protocol.types.Type.BOOLEAN;
import static org.apache.kafka.common.protocol.types.Type.INT16;
import static org.apache.kafka.common.protocol.types.Type.INT32;
import static org.apache.kafka.common.protocol.types.Type.INT64;

public class WriteTxnMarkersRequest extends AbstractRequest {
private static final String COORDINATOR_EPOCH_KEY_NAME = "coordinator_epoch";
private static final String TXN_MARKERS_KEY_NAME = "transaction_markers";

private static final String PRODUCER_ID_KEY_NAME = "producer_id";
private static final String PRODUCER_EPOCH_KEY_NAME = "producer_epoch";
private static final String TRANSACTION_RESULT_KEY_NAME = "transaction_result";
private static final String TOPICS_KEY_NAME = "topics";
private static final String PARTITIONS_KEY_NAME = "partitions";

private static final Schema WRITE_TXN_MARKERS_ENTRY_V0 = new Schema(
new Field(PRODUCER_ID_KEY_NAME, INT64, "Current producer id in use by the transactional id."),
new Field(PRODUCER_EPOCH_KEY_NAME, INT16, "Current epoch associated with the producer id."),
new Field(TRANSACTION_RESULT_KEY_NAME, BOOLEAN, "The result of the transaction to write to the " +
"partitions (false = ABORT, true = COMMIT)."),
new Field(TOPICS_KEY_NAME, new ArrayOf(new Schema(
TOPIC_NAME,
new Field(PARTITIONS_KEY_NAME, new ArrayOf(INT32)))), "The partitions to write markers for."),
new Field(COORDINATOR_EPOCH_KEY_NAME, INT32, "Epoch associated with the transaction state partition " +
"hosted by this transaction coordinator"));

private static final Schema WRITE_TXN_MARKERS_REQUEST_V0 = new Schema(
new Field(TXN_MARKERS_KEY_NAME, new ArrayOf(WRITE_TXN_MARKERS_ENTRY_V0), "The transaction markers to " +
"be written."));

public static Schema[] schemaVersions() {
return new Schema[]{WRITE_TXN_MARKERS_REQUEST_V0};
}

public static class TxnMarkerEntry {
private final long producerId;
Expand Down Expand Up @@ -106,16 +72,15 @@ public List<TopicPartition> partitions() {
return partitions;
}


@Override
public String toString() {
return "TxnMarkerEntry{" +
"producerId=" + producerId +
", producerEpoch=" + producerEpoch +
", coordinatorEpoch=" + coordinatorEpoch +
", result=" + result +
", partitions=" + partitions +
'}';
"producerId=" + producerId +
", producerEpoch=" + producerEpoch +
", coordinatorEpoch=" + coordinatorEpoch +
", result=" + result +
", partitions=" + partitions +
'}';
}

@Override
Expand All @@ -124,10 +89,10 @@ public boolean equals(final Object o) {
if (o == null || getClass() != o.getClass()) return false;
final TxnMarkerEntry that = (TxnMarkerEntry) o;
return producerId == that.producerId &&
producerEpoch == that.producerEpoch &&
coordinatorEpoch == that.coordinatorEpoch &&
result == that.result &&
Objects.equals(partitions, that.partitions);
producerEpoch == that.producerEpoch &&
coordinatorEpoch == that.coordinatorEpoch &&
result == that.result &&
Objects.equals(partitions, that.partitions);
}

@Override
Expand All @@ -137,106 +102,93 @@ public int hashCode() {
}

public static class Builder extends AbstractRequest.Builder<WriteTxnMarkersRequest> {
private final List<TxnMarkerEntry> markers;

public Builder(List<TxnMarkerEntry> markers) {
public final WriteTxnMarkersRequestData data;

public Builder(final List<TxnMarkerEntry> markers) {
super(ApiKeys.WRITE_TXN_MARKERS);
this.markers = markers;
List<WritableTxnMarker> dataMarkers = new ArrayList<>();
for (TxnMarkerEntry marker : markers) {
final Map<String, WritableTxnMarkerTopic> topicMap = new HashMap<>();
for (TopicPartition topicPartition : marker.partitions) {
WritableTxnMarkerTopic topic = topicMap.getOrDefault(topicPartition.topic(),
new WritableTxnMarkerTopic()
.setName(topicPartition.topic()));
topic.partitionIndexes().add(topicPartition.partition());
topicMap.put(topicPartition.topic(), topic);
}

dataMarkers.add(new WritableTxnMarker()
.setProducerId(marker.producerId)
.setProducerEpoch(marker.producerEpoch)
.setCoordinatorEpoch(marker.coordinatorEpoch)
.setTransactionResult(marker.transactionResult().id)
.setTopics(new ArrayList<>(topicMap.values())));
}
this.data = new WriteTxnMarkersRequestData().setMarkers(dataMarkers);
}

@Override
public WriteTxnMarkersRequest build(short version) {
return new WriteTxnMarkersRequest(version, markers);
return new WriteTxnMarkersRequest(data, version);
}
}

private final List<TxnMarkerEntry> markers;
public final WriteTxnMarkersRequestData data;

private WriteTxnMarkersRequest(short version, List<TxnMarkerEntry> markers) {
private WriteTxnMarkersRequest(WriteTxnMarkersRequestData data, short version) {
super(ApiKeys.WRITE_TXN_MARKERS, version);

this.markers = markers;
this.data = data;
}

public WriteTxnMarkersRequest(Struct struct, short version) {
super(ApiKeys.WRITE_TXN_MARKERS, version);
List<TxnMarkerEntry> markers = new ArrayList<>();
Object[] markersArray = struct.getArray(TXN_MARKERS_KEY_NAME);
for (Object markerObj : markersArray) {
Struct markerStruct = (Struct) markerObj;

long producerId = markerStruct.getLong(PRODUCER_ID_KEY_NAME);
short producerEpoch = markerStruct.getShort(PRODUCER_EPOCH_KEY_NAME);
int coordinatorEpoch = markerStruct.getInt(COORDINATOR_EPOCH_KEY_NAME);
TransactionResult result = TransactionResult.forId(markerStruct.getBoolean(TRANSACTION_RESULT_KEY_NAME));

List<TopicPartition> partitions = new ArrayList<>();
Object[] topicPartitionsArray = markerStruct.getArray(TOPICS_KEY_NAME);
for (Object topicPartitionObj : topicPartitionsArray) {
Struct topicPartitionStruct = (Struct) topicPartitionObj;
String topic = topicPartitionStruct.get(TOPIC_NAME);
for (Object partitionObj : topicPartitionStruct.getArray(PARTITIONS_KEY_NAME)) {
partitions.add(new TopicPartition(topic, (Integer) partitionObj));
}
}

markers.add(new TxnMarkerEntry(producerId, producerEpoch, coordinatorEpoch, result, partitions));
}

this.markers = markers;
}


public List<TxnMarkerEntry> markers() {
return markers;
this.data = new WriteTxnMarkersRequestData(struct, version);
}

@Override
protected Struct toStruct() {
Struct struct = new Struct(ApiKeys.WRITE_TXN_MARKERS.requestSchema(version()));

Object[] markersArray = new Object[markers.size()];
int i = 0;
for (TxnMarkerEntry entry : markers) {
Struct markerStruct = struct.instance(TXN_MARKERS_KEY_NAME);
markerStruct.set(PRODUCER_ID_KEY_NAME, entry.producerId);
markerStruct.set(PRODUCER_EPOCH_KEY_NAME, entry.producerEpoch);
markerStruct.set(COORDINATOR_EPOCH_KEY_NAME, entry.coordinatorEpoch);
markerStruct.set(TRANSACTION_RESULT_KEY_NAME, entry.result.id);

Map<String, List<Integer>> mappedPartitions = CollectionUtils.groupPartitionsByTopic(entry.partitions);
Object[] partitionsArray = new Object[mappedPartitions.size()];
int j = 0;
for (Map.Entry<String, List<Integer>> topicAndPartitions : mappedPartitions.entrySet()) {
Struct topicPartitionsStruct = markerStruct.instance(TOPICS_KEY_NAME);
topicPartitionsStruct.set(TOPIC_NAME, topicAndPartitions.getKey());
topicPartitionsStruct.set(PARTITIONS_KEY_NAME, topicAndPartitions.getValue().toArray());
partitionsArray[j++] = topicPartitionsStruct;
}
markerStruct.set(TOPICS_KEY_NAME, partitionsArray);
markersArray[i++] = markerStruct;
}
struct.set(TXN_MARKERS_KEY_NAME, markersArray);

return struct;
return data.toStruct(version());
}

@Override
public WriteTxnMarkersResponse getErrorResponse(int throttleTimeMs, Throwable e) {
Errors error = Errors.forException(e);

Map<Long, Map<TopicPartition, Errors>> errors = new HashMap<>(markers.size());
for (TxnMarkerEntry entry : markers) {
Map<TopicPartition, Errors> errorsPerPartition = new HashMap<>(entry.partitions.size());
for (TopicPartition partition : entry.partitions)
errorsPerPartition.put(partition, error);

errors.put(entry.producerId, errorsPerPartition);
final Map<Long, Map<TopicPartition, Errors>> errors = new HashMap<>(data.markers().size());
for (WritableTxnMarker markerEntry : data.markers()) {
Map<TopicPartition, Errors> errorsPerPartition = new HashMap<>();
for (WritableTxnMarkerTopic topic : markerEntry.topics()) {
for (Integer partitionIdx : topic.partitionIndexes()) {
errorsPerPartition.put(new TopicPartition(topic.name(), partitionIdx), error);
}
}
errors.put(markerEntry.producerId(), errorsPerPartition);
}

return new WriteTxnMarkersResponse(errors);
}

public List<TxnMarkerEntry> markers() {
List<TxnMarkerEntry> markers = new ArrayList<>();
for (WritableTxnMarker markerEntry : data.markers()) {
List<TopicPartition> topicPartitions = new ArrayList<>();
for (WritableTxnMarkerTopic topic : markerEntry.topics()) {
for (Integer partitionIdx : topic.partitionIndexes()) {
topicPartitions.add(new TopicPartition(topic.name(), partitionIdx));
}
}
markers.add(new TxnMarkerEntry(
markerEntry.producerId(),
markerEntry.producerEpoch(),
markerEntry.coordinatorEpoch(),
TransactionResult.forId(markerEntry.transactionResult()),
topicPartitions)
);
}
return markers;
}

public static WriteTxnMarkersRequest parse(ByteBuffer buffer, short version) {
return new WriteTxnMarkersRequest(ApiKeys.WRITE_TXN_MARKERS.parseRequest(version, buffer), version);
}
Expand All @@ -246,11 +198,11 @@ public boolean equals(final Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
final WriteTxnMarkersRequest that = (WriteTxnMarkersRequest) o;
return Objects.equals(markers, that.markers);
return Objects.equals(this.data, that.data);
}

@Override
public int hashCode() {
return Objects.hash(markers);
return Objects.hash(this.data);
}
}
Loading