From 9fefdcfc7be75c247889dd2dfdddff5f3a1917bf Mon Sep 17 00:00:00 2001 From: abbccdda Date: Thu, 4 Jul 2019 23:41:46 -0700 Subject: [PATCH 1/2] add txn marker auto --- .../apache/kafka/common/protocol/ApiKeys.java | 8 +- .../common/requests/AbstractResponse.java | 2 +- .../requests/WriteTxnMarkersRequest.java | 192 +++++++----------- .../requests/WriteTxnMarkersResponse.java | 164 ++++++--------- .../requests/WriteTxnMarkersRequestTest.java | 81 ++++++++ .../requests/WriteTxnMarkersResponseTest.java | 60 ++++++ .../internals/StateDirectoryTest.java | 1 + 7 files changed, 278 insertions(+), 230 deletions(-) create mode 100644 clients/src/test/java/org/apache/kafka/common/requests/WriteTxnMarkersRequestTest.java create mode 100644 clients/src/test/java/org/apache/kafka/common/requests/WriteTxnMarkersResponseTest.java diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java b/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java index 6549d8fbb12d3..e96917008485f 100644 --- a/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java +++ b/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java @@ -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; @@ -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; @@ -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), diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java index 9d3f39d4f5124..301a4fb5eb3c3 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java @@ -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: diff --git a/clients/src/main/java/org/apache/kafka/common/requests/WriteTxnMarkersRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/WriteTxnMarkersRequest.java index 33f9bb586d9a5..c27263384a5ee 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/WriteTxnMarkersRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/WriteTxnMarkersRequest.java @@ -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; @@ -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; @@ -106,16 +72,15 @@ public List 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 @@ -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 @@ -137,106 +102,93 @@ public int hashCode() { } public static class Builder extends AbstractRequest.Builder { - private final List markers; - public Builder(List markers) { + public final WriteTxnMarkersRequestData data; + + public Builder(final List markers) { super(ApiKeys.WRITE_TXN_MARKERS); - this.markers = markers; + List dataMarkers = new ArrayList<>(); + for (TxnMarkerEntry marker : markers) { + final Map 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 markers; + public final WriteTxnMarkersRequestData data; - private WriteTxnMarkersRequest(short version, List 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 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 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 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> mappedPartitions = CollectionUtils.groupPartitionsByTopic(entry.partitions); - Object[] partitionsArray = new Object[mappedPartitions.size()]; - int j = 0; - for (Map.Entry> 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> errors = new HashMap<>(markers.size()); - for (TxnMarkerEntry entry : markers) { - Map errorsPerPartition = new HashMap<>(entry.partitions.size()); - for (TopicPartition partition : entry.partitions) - errorsPerPartition.put(partition, error); - - errors.put(entry.producerId, errorsPerPartition); + final Map> errors = new HashMap<>(data.markers().size()); + for (WritableTxnMarker markerEntry : data.markers()) { + Map 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 markers() { + List markers = new ArrayList<>(); + for (WritableTxnMarker markerEntry : data.markers()) { + List 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); } @@ -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); } } diff --git a/clients/src/main/java/org/apache/kafka/common/requests/WriteTxnMarkersResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/WriteTxnMarkersResponse.java index 92b5fc0b29dc2..3fde41a99f144 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/WriteTxnMarkersResponse.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/WriteTxnMarkersResponse.java @@ -17,133 +17,87 @@ package org.apache.kafka.common.requests; import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.message.WriteTxnMarkersResponseData; +import org.apache.kafka.common.message.WriteTxnMarkersResponseData.WritableTxnMarkerPartitionResult; +import org.apache.kafka.common.message.WriteTxnMarkersResponseData.WritableTxnMarkerResult; +import org.apache.kafka.common.message.WriteTxnMarkersResponseData.WritableTxnMarkerTopicResult; 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; import java.util.HashMap; +import java.util.List; import java.util.Map; -import static org.apache.kafka.common.protocol.CommonFields.ERROR_CODE; -import static org.apache.kafka.common.protocol.CommonFields.PARTITION_ID; -import static org.apache.kafka.common.protocol.CommonFields.TOPIC_NAME; -import static org.apache.kafka.common.protocol.types.Type.INT64; - +/** + * Possible error codes: + * + * - {@link Errors#CORRUPT_MESSAGE} + * - {@link Errors#INVALID_PRODUCER_EPOCH} + * - {@link Errors#UNKNOWN_TOPIC_OR_PARTITION} + * - {@link Errors#NOT_LEADER_FOR_PARTITION} + * - {@link Errors#MESSAGE_TOO_LARGE} + * - {@link Errors#RECORD_LIST_TOO_LARGE} + * - {@link Errors#NOT_ENOUGH_REPLICAS} + * - {@link Errors#NOT_ENOUGH_REPLICAS_AFTER_APPEND} + * - {@link Errors#INVALID_REQUIRED_ACKS} + * - {@link Errors#TRANSACTION_COORDINATOR_FENCED} + * - {@link Errors#REQUEST_TIMED_OUT} + * - {@link Errors#CLUSTER_AUTHORIZATION_FAILED} + */ public class WriteTxnMarkersResponse extends AbstractResponse { - private static final String TXN_MARKERS_KEY_NAME = "transaction_markers"; - - private static final String PRODUCER_ID_KEY_NAME = "producer_id"; - private static final String TOPICS_KEY_NAME = "topics"; - private static final String PARTITIONS_KEY_NAME = "partitions"; - - private static final Schema WRITE_TXN_MARKERS_PARTITION_ERROR_RESPONSE_V0 = new Schema( - PARTITION_ID, - ERROR_CODE); - - 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(TOPICS_KEY_NAME, new ArrayOf(new Schema( - TOPIC_NAME, - new Field(PARTITIONS_KEY_NAME, new ArrayOf(WRITE_TXN_MARKERS_PARTITION_ERROR_RESPONSE_V0)))), - "Errors per partition from writing markers.")); - - private static final Schema WRITE_TXN_MARKERS_RESPONSE_V0 = new Schema( - new Field(TXN_MARKERS_KEY_NAME, new ArrayOf(WRITE_TXN_MARKERS_ENTRY_V0), "Errors per partition from " + - "writing markers.")); - - public static Schema[] schemaVersions() { - return new Schema[]{WRITE_TXN_MARKERS_RESPONSE_V0}; - } - - // Possible error codes: - // CorruptRecord - // InvalidProducerEpoch - // UnknownTopicOrPartition - // NotLeaderForPartition - // MessageTooLarge - // RecordListTooLarge - // NotEnoughReplicas - // NotEnoughReplicasAfterAppend - // InvalidRequiredAcks - // TransactionCoordinatorFenced - // RequestTimeout - // ClusterAuthorizationFailed private final Map> errors; + public final WriteTxnMarkersResponseData data; public WriteTxnMarkersResponse(Map> errors) { - this.errors = errors; - } - - public WriteTxnMarkersResponse(Struct struct) { - Map> errors = new HashMap<>(); - - Object[] responseArray = struct.getArray(TXN_MARKERS_KEY_NAME); - for (Object responseObj : responseArray) { - Struct responseStruct = (Struct) responseObj; - - long producerId = responseStruct.getLong(PRODUCER_ID_KEY_NAME); - - Map errorPerPartition = new HashMap<>(); - Object[] topicPartitionsArray = responseStruct.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)) { - Struct partitionStruct = (Struct) partitionObj; - Integer partition = partitionStruct.get(PARTITION_ID); - Errors error = Errors.forCode(partitionStruct.get(ERROR_CODE)); - errorPerPartition.put(new TopicPartition(topic, partition), error); - } + List markers = new ArrayList<>(); + for (Map.Entry> markerEntry : errors.entrySet()) { + Map responseTopicDataMap = new HashMap<>(); + for (Map.Entry topicEntry : markerEntry.getValue().entrySet()) { + TopicPartition topicPartition = topicEntry.getKey(); + String topicName = topicPartition.topic(); + + WritableTxnMarkerTopicResult topic = + responseTopicDataMap.getOrDefault(topicName, new WritableTxnMarkerTopicResult().setName(topicName)); + topic.partitions().add(new WritableTxnMarkerPartitionResult() + .setErrorCode(topicEntry.getValue().code()) + .setPartitionIndex(topicPartition.partition()) + ); + responseTopicDataMap.put(topicName, topic); } - errors.put(producerId, errorPerPartition); + + markers.add(new WritableTxnMarkerResult() + .setProducerId(markerEntry.getKey()) + .setTopics(new ArrayList<>(responseTopicDataMap.values())) + ); } this.errors = errors; + this.data = new WriteTxnMarkersResponseData() + .setMarkers(markers); } - @Override - protected Struct toStruct(short version) { - Struct struct = new Struct(ApiKeys.WRITE_TXN_MARKERS.responseSchema(version)); - - Object[] responsesArray = new Object[errors.size()]; - int k = 0; - for (Map.Entry> responseEntry : errors.entrySet()) { - Struct responseStruct = struct.instance(TXN_MARKERS_KEY_NAME); - responseStruct.set(PRODUCER_ID_KEY_NAME, responseEntry.getKey()); - - Map partitionAndErrors = responseEntry.getValue(); - Map> mappedPartitions = CollectionUtils.groupPartitionDataByTopic(partitionAndErrors); - Object[] partitionsArray = new Object[mappedPartitions.size()]; - int i = 0; - for (Map.Entry> topicAndPartitions : mappedPartitions.entrySet()) { - Struct topicPartitionsStruct = responseStruct.instance(TOPICS_KEY_NAME); - topicPartitionsStruct.set(TOPIC_NAME, topicAndPartitions.getKey()); - Map partitionIdAndErrors = topicAndPartitions.getValue(); - - Object[] partitionAndErrorsArray = new Object[partitionIdAndErrors.size()]; - int j = 0; - for (Map.Entry partitionAndError : partitionIdAndErrors.entrySet()) { - Struct partitionAndErrorStruct = topicPartitionsStruct.instance(PARTITIONS_KEY_NAME); - partitionAndErrorStruct.set(PARTITION_ID, partitionAndError.getKey()); - partitionAndErrorStruct.set(ERROR_CODE, partitionAndError.getValue().code()); - partitionAndErrorsArray[j++] = partitionAndErrorStruct; + public WriteTxnMarkersResponse(Struct struct, short version) { + this.data = new WriteTxnMarkersResponseData(struct, version); + this.errors = new HashMap<>(); + for (WritableTxnMarkerResult marker : data.markers()) { + Map topicPartitionErrorsMap = new HashMap<>(); + for (WritableTxnMarkerTopicResult topic: marker.topics()) { + for (WritableTxnMarkerPartitionResult partitionResult: topic.partitions()) { + topicPartitionErrorsMap.put(new TopicPartition(topic.name(), partitionResult.partitionIndex()), + Errors.forCode(partitionResult.errorCode())); } - topicPartitionsStruct.set(PARTITIONS_KEY_NAME, partitionAndErrorsArray); - partitionsArray[i++] = topicPartitionsStruct; } - responseStruct.set(TOPICS_KEY_NAME, partitionsArray); - - responsesArray[k++] = responseStruct; + errors.put(marker.producerId(), topicPartitionErrorsMap); } + } - struct.set(TXN_MARKERS_KEY_NAME, responsesArray); - return struct; + @Override + protected Struct toStruct(short version) { + return data.toStruct(version); } public Map errors(long producerId) { @@ -161,6 +115,6 @@ public Map errorCounts() { } public static WriteTxnMarkersResponse parse(ByteBuffer buffer, short version) { - return new WriteTxnMarkersResponse(ApiKeys.WRITE_TXN_MARKERS.parseResponse(version, buffer)); + return new WriteTxnMarkersResponse(ApiKeys.WRITE_TXN_MARKERS.parseResponse(version, buffer), version); } } diff --git a/clients/src/test/java/org/apache/kafka/common/requests/WriteTxnMarkersRequestTest.java b/clients/src/test/java/org/apache/kafka/common/requests/WriteTxnMarkersRequestTest.java new file mode 100644 index 0000000000000..7273e4734cb4c --- /dev/null +++ b/clients/src/test/java/org/apache/kafka/common/requests/WriteTxnMarkersRequestTest.java @@ -0,0 +1,81 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.common.requests; + +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.common.protocol.Errors; +import org.junit.Before; +import org.junit.Test; + +import java.util.Collections; +import java.util.List; + +import static org.junit.Assert.assertEquals; + +public class WriteTxnMarkersRequestTest { + + private static long producerId = 10L; + private static short producerEpoch = 2; + private static int coordinatorEpoch = 1; + private static TransactionResult result = TransactionResult.COMMIT; + private static TopicPartition topicPartition = new TopicPartition("topic", 73); + + protected static int throttleTimeMs = 10; + + private static List markers; + + @Before + public void setUp() { + markers = Collections.singletonList( + new WriteTxnMarkersRequest.TxnMarkerEntry( + producerId, producerEpoch, coordinatorEpoch, + result, Collections.singletonList(topicPartition)) + ); + } + + @Test + public void testConstructor() { + WriteTxnMarkersRequest.Builder builder = new WriteTxnMarkersRequest.Builder(markers); + for (short version = 0; version <= ApiKeys.WRITE_TXN_MARKERS.latestVersion(); version++) { + WriteTxnMarkersRequest request = builder.build(version); + assertEquals(1, request.markers().size()); + WriteTxnMarkersRequest.TxnMarkerEntry marker = request.markers().get(0); + assertEquals(producerId, marker.producerId()); + assertEquals(producerEpoch, marker.producerEpoch()); + assertEquals(coordinatorEpoch, marker.coordinatorEpoch()); + assertEquals(result, marker.transactionResult()); + assertEquals(Collections.singletonList(topicPartition), marker.partitions()); + } + } + + @Test + public void testGetErrorResponse() { + WriteTxnMarkersRequest.Builder builder = new WriteTxnMarkersRequest.Builder(markers); + for (short version = 0; version <= ApiKeys.WRITE_TXN_MARKERS.latestVersion(); version++) { + WriteTxnMarkersRequest request = builder.build(version); + WriteTxnMarkersResponse errorResponse = + request.getErrorResponse(throttleTimeMs, Errors.UNKNOWN_PRODUCER_ID.exception()); + + assertEquals(Collections.singletonMap( + topicPartition, Errors.UNKNOWN_PRODUCER_ID), errorResponse.errors(producerId)); + assertEquals(Collections.singletonMap(Errors.UNKNOWN_PRODUCER_ID, 1), errorResponse.errorCounts()); + // Write txn marker has no throttle time defined in response. + assertEquals(0, errorResponse.throttleTimeMs()); + } + } +} diff --git a/clients/src/test/java/org/apache/kafka/common/requests/WriteTxnMarkersResponseTest.java b/clients/src/test/java/org/apache/kafka/common/requests/WriteTxnMarkersResponseTest.java new file mode 100644 index 0000000000000..4cec88c22e923 --- /dev/null +++ b/clients/src/test/java/org/apache/kafka/common/requests/WriteTxnMarkersResponseTest.java @@ -0,0 +1,60 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.common.requests; + +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.protocol.Errors; +import org.junit.Before; +import org.junit.Test; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +import static org.junit.Assert.assertEquals; + +public class WriteTxnMarkersResponseTest { + + private static long producerIdOne = 1L; + private static long producerIdTwo = 2L; + + private static TopicPartition tp1 = new TopicPartition("topic", 1); + private static TopicPartition tp2 = new TopicPartition("topic", 2); + + private static Errors pidOneError = Errors.UNKNOWN_PRODUCER_ID; + private static Errors pidTwoError = Errors.INVALID_PRODUCER_EPOCH; + + private static Map> errorMap; + + @Before + public void setUp() { + errorMap = new HashMap<>(); + errorMap.put(producerIdOne, Collections.singletonMap(tp1, pidOneError)); + errorMap.put(producerIdTwo, Collections.singletonMap(tp2, pidTwoError)); + } + @Test + public void testConstructorWithStruct() { + + Map expectedErrorCounts = new HashMap<>(); + expectedErrorCounts.put(Errors.UNKNOWN_PRODUCER_ID, 1); + expectedErrorCounts.put(Errors.INVALID_PRODUCER_EPOCH, 1); + WriteTxnMarkersResponse response = new WriteTxnMarkersResponse(errorMap); + assertEquals(expectedErrorCounts, response.errorCounts()); + assertEquals(Collections.singletonMap(tp1, pidOneError), response.errors(producerIdOne)); + assertEquals(Collections.singletonMap(tp2, pidTwoError), response.errors(producerIdTwo)); + } +} diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java index 245675555bc72..641c6a81cbea6 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java @@ -270,6 +270,7 @@ public void shouldCleanUpTaskStateDirectoriesThatAreNotCurrentlyLocked() throws files = Arrays.stream( Objects.requireNonNull(directory.listNonEmptyTaskDirectories())).collect(Collectors.toSet()); + assertEquals(mkSet(dir0, dir1), files); } finally { directory.unlock(task0); From 7d087fd880503a256b6e6293659a1049789ab6cd Mon Sep 17 00:00:00 2001 From: Mickael Maison Date: Thu, 19 Mar 2020 13:37:10 +0000 Subject: [PATCH 2/2] Remove extra empty line --- .../kafka/streams/processor/internals/StateDirectoryTest.java | 1 - 1 file changed, 1 deletion(-) diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java index 641c6a81cbea6..245675555bc72 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java @@ -270,7 +270,6 @@ public void shouldCleanUpTaskStateDirectoriesThatAreNotCurrentlyLocked() throws files = Arrays.stream( Objects.requireNonNull(directory.listNonEmptyTaskDirectories())).collect(Collectors.toSet()); - assertEquals(mkSet(dir0, dir1), files); } finally { directory.unlock(task0);