You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "sjhajharia (via GitHub)" <gi...@apache.org> on 2024/03/07 07:25:24 UTC
[PR] [KIP-890] Introducing the AbortableTransactionException [kafka]
sjhajharia opened a new pull request, #15486:
URL: https://github.com/apache/kafka/pull/15486
As a part of KIP-890, we are introducing a new class of Exceptions which when encountered shall lead to Aborting the ongoing Transaction. The following PR introduces the same with client side handling and server side changes.
On client Side, the code attempts to handle the exception as an Abortable error and ensure that it doesn't take the producer to a fatal state
On the server side, we have bumped the ProduceRequest and ProduceResponse Version. And then added the necessary handling for the same maintaining the backward compatibility with clients.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1523603267
##########
core/src/main/scala/kafka/server/AddPartitionsToTxnManager.scala:
##########
@@ -42,14 +42,26 @@ object AddPartitionsToTxnManager {
val VerificationTimeMsMetricName = "VerificationTimeMs"
}
+/**
+ * This is an enum which handles the Partition Response based on the Produce Request Version and the exact operation
+ * defaultOperation: This is the default workflow which maps to cases when the Produce Request Version was lower than expected or when exercising the offset commit request path
Review Comment:
I think we should include this change in the offset commit path. It will require bumping that request version as well.
> We will bump the ProduceRequest/Response and TxnOffsetCommitRequest/Response version to indicate the client is using the new protocol that doesnβt require adding partitions to transactions and will implicitly do so. The bump will also support new errors ABORTABLE_ERROR
We are just doing 2 bumps for the above comment
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on PR #15486:
URL: https://github.com/apache/kafka/pull/15486#issuecomment-2003011892
@jolshan, I updated the PR addressing the comments you left. I have renamed the enum to `ApiVersionErrorMapper`. Please do let me know if that looks better.
I will also wait for your confirmation on whether we need any changes in the exception naming and if all the Transactional APIs need to be bumped.
Thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1518061441
##########
clients/src/main/java/org/apache/kafka/common/errors/AbortableTransactionException.java:
##########
@@ -0,0 +1,23 @@
+/*
+ * 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.errors;
+
+public class AbortableTransactionException extends ApiException {
Review Comment:
I know there were some questions about the naming -- this seems reasonable but just wanted to call it out.
cc: @artemlivshits
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "CalvinConfluent (via GitHub)" <gi...@apache.org>.
CalvinConfluent commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1523705762
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -1504,6 +1507,8 @@ public void handleResponse(AbstractResponse response) {
fatalError(error.exception());
} else if (error == Errors.GROUP_AUTHORIZATION_FAILED) {
abortableError(GroupAuthorizationException.forGroupId(key));
+ } else if (error == Errors.ABORTABLE_TRANSACTION_EXCEPTION) {
Review Comment:
Does FindCoordinatorRequest also return the ABORTABLE_TRANSACTION_EXCEPTION?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1529793846
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java:
##########
@@ -3523,6 +3524,128 @@ public void testForegroundInvalidStateTransitionIsRecoverable() {
assertFalse(transactionManager.hasOngoingTransaction());
}
+ @Test
+ public void testAbortableTransactionExceptionInInitProducerId() {
+ TransactionalRequestResult initPidResult = transactionManager.initializeTransactions();
+ prepareFindCoordinatorResponse(Errors.NONE, false, CoordinatorType.TRANSACTION, transactionalId);
+ runUntil(() -> transactionManager.coordinator(CoordinatorType.TRANSACTION) != null);
+ assertEquals(brokerNode, transactionManager.coordinator(CoordinatorType.TRANSACTION));
+
+ prepareInitPidResponse(Errors.ABORTABLE_TRANSACTION, false, producerId, RecordBatch.NO_PRODUCER_EPOCH);
+ runUntil(transactionManager::hasError);
+ assertTrue(initPidResult.isCompleted());
+ assertFalse(initPidResult.isSuccessful());
+ assertThrows(AbortableTransactionException.class, initPidResult::await);
+ assertAbortableError(AbortableTransactionException.class);
+ }
+
+ @Test
+ public void testAbortableTransactionExceptionInAddPartitions() {
+ final TopicPartition tp = new TopicPartition("foo", 0);
+
+ doInitTransactions();
+
+ transactionManager.beginTransaction();
+ transactionManager.maybeAddPartition(tp);
+
+ prepareAddPartitionsToTxn(tp, Errors.ABORTABLE_TRANSACTION);
+ runUntil(transactionManager::hasError);
+ assertTrue(transactionManager.lastError() instanceof AbortableTransactionException);
+
+ assertAbortableError(AbortableTransactionException.class);
+ }
+
+ @Test
+ public void testAbortableTransactionExceptionInFindCoordinator() {
+ doInitTransactions();
+
+ transactionManager.beginTransaction();
+ TransactionalRequestResult sendOffsetsResult = transactionManager.sendOffsetsToTransaction(
+ singletonMap(new TopicPartition("foo", 0), new OffsetAndMetadata(39L)), new ConsumerGroupMetadata(consumerGroupId));
+
+ prepareAddOffsetsToTxnResponse(Errors.NONE, consumerGroupId, producerId, epoch);
+ runUntil(() -> !transactionManager.hasPartitionsToAdd());
+
+ prepareFindCoordinatorResponse(Errors.ABORTABLE_TRANSACTION, false, CoordinatorType.GROUP, consumerGroupId);
+ runUntil(transactionManager::hasError);
+ assertTrue(transactionManager.lastError() instanceof AbortableTransactionException);
+
+ runUntil(sendOffsetsResult::isCompleted);
+ assertFalse(sendOffsetsResult.isSuccessful());
+ assertTrue(sendOffsetsResult.error() instanceof AbortableTransactionException);
+
+ assertAbortableError(AbortableTransactionException.class);
+ }
+
+ @Test
+ public void testAbortableTransactionExceptionInEndTxn() throws InterruptedException {
+ doInitTransactions();
+
+ transactionManager.beginTransaction();
+ transactionManager.maybeAddPartition(tp0);
+ TransactionalRequestResult commitResult = transactionManager.beginCommit();
+
+ Future<RecordMetadata> responseFuture = appendToAccumulator(tp0);
+
+ assertFalse(responseFuture.isDone());
+ prepareAddPartitionsToTxnResponse(Errors.NONE, tp0, epoch, producerId);
+ prepareProduceResponse(Errors.NONE, producerId, epoch);
+ prepareEndTxnResponse(Errors.ABORTABLE_TRANSACTION, TransactionResult.COMMIT, producerId, epoch);
+
+ runUntil(commitResult::isCompleted);
+ runUntil(responseFuture::isDone);
+
+ assertThrows(KafkaException.class, commitResult::await);
+ assertFalse(commitResult.isSuccessful());
+ assertTrue(commitResult.isAcked());
+
+ assertAbortableError(AbortableTransactionException.class);
+ // make sure the exception was thrown directly from the follow-up calls.
Review Comment:
Fixed this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1526861466
##########
clients/src/main/java/org/apache/kafka/common/protocol/Errors.java:
##########
@@ -392,7 +393,8 @@ public enum Errors {
UNKNOWN_CONTROLLER_ID(116, "This controller ID is not known.", UnknownControllerIdException::new),
UNKNOWN_SUBSCRIPTION_ID(117, "Client sent a push telemetry request with an invalid or outdated subscription ID.", UnknownSubscriptionIdException::new),
TELEMETRY_TOO_LARGE(118, "Client sent a push telemetry request larger than the maximum size the broker will accept.", TelemetryTooLargeException::new),
- INVALID_REGISTRATION(119, "The controller has considered the broker registration to be invalid.", InvalidRegistrationException::new);
+ INVALID_REGISTRATION(119, "The controller has considered the broker registration to be invalid.", InvalidRegistrationException::new),
+ ABORTABLE_TRANSACTION(120, "This maps to all errors which can be fixed by aborting the current transaction.", AbortableTransactionException::new);
Review Comment:
Minor nit here with the comment. Maybe we can say -- something like
"The server encountered an error with the transaction. The client can abort the transaction to continue using this transactional ID."
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1519333893
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1007,7 +1010,8 @@ class ReplicaManager(val config: KafkaConfig,
transactionalId,
producerId,
producerEpoch,
- generalizedCallback
+ generalizedCallback,
+ ApiKeys.PRODUCE.latestVersion
Review Comment:
I don't think that including the latest Produce Request Version will break any code flow. So should we just leave a comment denoting the same?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1518063503
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -3151,6 +3152,45 @@ public void testInvalidTxnStateIsAnAbortableError() throws Exception {
txnManager.beginTransaction();
}
+
+ @Test
+ public void testAbortableTxnExceptionIsAnAbortableError() throws Exception {
+ ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(123456L, (short) 0);
+ apiVersions.update("0", NodeApiVersions.create(ApiKeys.INIT_PRODUCER_ID.id, (short) 0, (short) 3));
+ TransactionManager txnManager = new TransactionManager(logContext, "textAbortableTxnException", 60000, 100, apiVersions);
+
+ setupWithTransactionState(txnManager);
+ doInitTransactions(txnManager, producerIdAndEpoch);
+
+ txnManager.beginTransaction();
+ txnManager.maybeAddPartition(tp0);
+ client.prepareResponse(buildAddPartitionsToTxnResponseData(0, Collections.singletonMap(tp0, Errors.NONE)));
+ sender.runOnce();
+
+ Future<RecordMetadata> request = appendToAccumulator(tp0);
+ sender.runOnce(); // send request
+ sendIdempotentProducerResponse(0, tp0, Errors.ABORTABLE_TRANSACTION_EXCEPTION, -1);
+
+ // Return InvalidTxnState error. It should be abortable.
Review Comment:
Is this comment correct?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1521679508
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1007,7 +1010,8 @@ class ReplicaManager(val config: KafkaConfig,
transactionalId,
producerId,
producerEpoch,
- generalizedCallback
+ generalizedCallback,
+ ApiKeys.PRODUCE.latestVersion
Review Comment:
What if we send a offset commit request that can't yet handle the error.
I think we may need a different data structure. Maybe something like enums to handle the various versions
0 -- default, 1 -- new error, 2 -- add partitions rather than verifying.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1526864714
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java:
##########
@@ -3523,6 +3524,129 @@ public void testForegroundInvalidStateTransitionIsRecoverable() {
assertFalse(transactionManager.hasOngoingTransaction());
}
+ @Test
+ public void testAbortableTransactionExceptionInInitProducerId() {
+ TransactionalRequestResult initPidResult = transactionManager.initializeTransactions();
+ prepareFindCoordinatorResponse(Errors.NONE, false, CoordinatorType.TRANSACTION, transactionalId);
+ runUntil(() -> transactionManager.coordinator(CoordinatorType.TRANSACTION) != null);
+ assertEquals(brokerNode, transactionManager.coordinator(CoordinatorType.TRANSACTION));
+
+ prepareInitPidResponse(Errors.ABORTABLE_TRANSACTION, false, producerId, RecordBatch.NO_PRODUCER_EPOCH);
+ runUntil(transactionManager::hasError);
+ assertTrue(initPidResult.isCompleted());
+ assertFalse(initPidResult.isSuccessful());
+ assertThrows(AbortableTransactionException.class, initPidResult::await);
+ assertAbortableError(AbortableTransactionException.class);
+ }
+
+ @Test
+ public void testAbortableTransactionExceptionInAddPartitions() {
+ final TopicPartition tp = new TopicPartition("foo", 0);
+
+ doInitTransactions();
+
+ transactionManager.beginTransaction();
+ transactionManager.maybeAddPartition(tp);
+
+ prepareAddPartitionsToTxn(tp, Errors.ABORTABLE_TRANSACTION);
+ runUntil(transactionManager::hasError);
+ assertTrue(transactionManager.lastError() instanceof AbortableTransactionException);
+
+ assertAbortableError(AbortableTransactionException.class);
+ }
+
+ @Test
+ public void testAbortableTransactionExceptionInFindCoordinator() {
+ doInitTransactions();
+
+ transactionManager.beginTransaction();
+ TransactionalRequestResult sendOffsetsResult = transactionManager.sendOffsetsToTransaction(
+ singletonMap(new TopicPartition("foo", 0), new OffsetAndMetadata(39L)), new ConsumerGroupMetadata(consumerGroupId));
+
+ prepareAddOffsetsToTxnResponse(Errors.NONE, consumerGroupId, producerId, epoch);
+ runUntil(() -> !transactionManager.hasPartitionsToAdd());
+
+ prepareFindCoordinatorResponse(Errors.ABORTABLE_TRANSACTION, false, CoordinatorType.GROUP, consumerGroupId);
+ runUntil(transactionManager::hasError);
+ assertTrue(transactionManager.lastError() instanceof AbortableTransactionException);
+
+ runUntil(sendOffsetsResult::isCompleted);
+ assertFalse(sendOffsetsResult.isSuccessful());
+ assertTrue(sendOffsetsResult.error() instanceof AbortableTransactionException);
+
+ assertAbortableError(AbortableTransactionException.class);
+ }
+
+ @Test
+ public void testAbortableTransactionExceptionInEndTxn() throws InterruptedException {
+ doInitTransactions();
+
+ transactionManager.beginTransaction();
+ transactionManager.maybeAddPartition(tp0);
+ TransactionalRequestResult commitResult = transactionManager.beginCommit();
+
+ Future<RecordMetadata> responseFuture = appendToAccumulator(tp0);
+
+ assertFalse(responseFuture.isDone());
+ prepareAddPartitionsToTxnResponse(Errors.NONE, tp0, epoch, producerId);
+ prepareProduceResponse(Errors.NONE, producerId, epoch);
+ prepareEndTxnResponse(Errors.ABORTABLE_TRANSACTION, TransactionResult.COMMIT, producerId, epoch);
+
+ runUntil(commitResult::isCompleted);
+ runUntil(responseFuture::isDone);
+
+ assertThrows(KafkaException.class, commitResult::await);
+ assertFalse(commitResult.isSuccessful());
+ assertTrue(commitResult.isAcked());
+
+ assertAbortableError(AbortableTransactionException.class);
+ // make sure the exception was thrown directly from the follow-up calls.
+
+ }
+
Review Comment:
nit: extra new line here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #15486:
URL: https://github.com/apache/kafka/pull/15486#issuecomment-2004546341
Hey @sjhajharia thanks for the updates. I think `ApiVersionErrorMapper` is also a bit tricky since the addPartitions change will not be related to errors at all. That's why I was thinking of something a bit more generic. "Operation" was probably the right direction rather than "errors". Sorry this part is a bit nuanced. Maybe something like "supportedOperation" and the comment could be something like the supported operation based on the client's request API version?
And yes, I think we should bump all the API versions to support this new error. I can update the KIP.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1536692361
##########
core/src/main/scala/kafka/server/AddPartitionsToTxnManager.scala:
##########
@@ -42,14 +42,26 @@ object AddPartitionsToTxnManager {
val VerificationTimeMsMetricName = "VerificationTimeMs"
}
+/**
+ * This is an enum which handles the Partition Response based on the Request Version and the exact operation
+ * defaultError: This is the default workflow which maps to cases when the Produce Request Version or the Txn_offset_commit request was lower than the first version supporting the new Error Class
+ * genericError: This maps to the case when the clients are updated to handle the AbortableTxnException
Review Comment:
I think it's not `AbortableTxnException`. Probably ought to give it its proper name.
##########
clients/src/main/java/org/apache/kafka/common/errors/AbortableTransactionException.java:
##########
@@ -0,0 +1,23 @@
+/*
+ * 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.errors;
+
+public class AbortableTransactionException extends ApiException {
Review Comment:
I tend to agree that `TransactionAbortableException` is very slightly more satisfying than `AbortableTransactionException`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1523608929
##########
core/src/main/scala/kafka/server/AddPartitionsToTxnManager.scala:
##########
@@ -42,14 +42,26 @@ object AddPartitionsToTxnManager {
val VerificationTimeMsMetricName = "VerificationTimeMs"
}
+/**
+ * This is an enum which handles the Partition Response based on the Produce Request Version and the exact operation
+ * defaultOperation: This is the default workflow which maps to cases when the Produce Request Version was lower than expected or when exercising the offset commit request path
Review Comment:
It is a similar process to pass in the the value, but we will need it for both group coordinator paths. Let me know if you need help.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1523717757
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -1504,6 +1507,8 @@ public void handleResponse(AbstractResponse response) {
fatalError(error.exception());
} else if (error == Errors.GROUP_AUTHORIZATION_FAILED) {
abortableError(GroupAuthorizationException.forGroupId(key));
+ } else if (error == Errors.ABORTABLE_TRANSACTION_EXCEPTION) {
Review Comment:
The idea of this error is that we are future proofing. In the future if we want the producer to abort the transaction (say we have a new use case that requires this behavior) we can rely on this error to do the correct thing for older clients.
We ran into this issue a lot when picking error codes for kip-890 part 1. If we had such an error then, we could have used it for the old clients. Instead, we chosen INVALID_TXN_STATE which has inconsistent behavior across clients.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1526868437
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -975,12 +978,13 @@ class ReplicaManager(val config: KafkaConfig,
/**
*
- * @param topicPartition the topic partition to maybe verify
- * @param transactionalId the transactional id for the transaction
- * @param producerId the producer id for the producer writing to the transaction
- * @param producerEpoch the epoch of the producer writing to the transaction
- * @param baseSequence the base sequence of the first record in the batch we are trying to append
- * @param callback the method to execute once the verification is either completed or returns an error
+ * @param topicPartition the topic partition to maybe verify
+ * @param transactionalId the transactional id for the transaction
+ * @param producerId the producer id for the producer writing to the transaction
+ * @param producerEpoch the epoch of the producer writing to the transaction
+ * @param baseSequence the base sequence of the first record in the batch we are trying to append
+ * @param callback the method to execute once the verification is either completed or returns an error
+ * @param partitionOperation the enum value which determines whether the client is ready to handle the new Abortable Txn Errors
Review Comment:
nit: to make this more generic we should say something like "determines which version of the client sent the request so the addPartitionsToTxnManager handles it appropriately".
I also wonder if we want to rename the argument. PartitionOperation is a little unclear.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1518064096
##########
core/src/main/scala/kafka/server/AddPartitionsToTxnManager.scala:
##########
@@ -228,11 +232,17 @@ class AddPartitionsToTxnManager(
val tp = new TopicPartition(topicResult.name, partitionResult.partitionIndex)
if (partitionResult.partitionErrorCode != Errors.NONE.code) {
// Producers expect to handle INVALID_PRODUCER_EPOCH in this scenario.
- val code =
+ var code =
if (partitionResult.partitionErrorCode == Errors.PRODUCER_FENCED.code)
Errors.INVALID_PRODUCER_EPOCH.code
else
partitionResult.partitionErrorCode
+ // For backward compatibility with clients.
Review Comment:
nice π
I think we can include this in val code as a subcase so we don't need the var
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on PR #15486:
URL: https://github.com/apache/kafka/pull/15486#issuecomment-2008655935
On that note @jolshan , do we any more open questions to address or we are good to go with this PR? Or are we in flux for [this](https://github.com/apache/kafka/pull/15486#discussion_r1523597267)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1526869018
##########
core/src/test/scala/unit/kafka/server/AddPartitionsToTxnManagerTest.scala:
##########
@@ -72,6 +72,8 @@ class AddPartitionsToTxnManagerTest {
private val versionMismatchResponse = clientResponse(null, mismatchException = new UnsupportedVersionException(""))
private val disconnectedResponse = clientResponse(null, disconnected = true)
+ //private val highestProduceVersion = ApiKeys.PRODUCE.oldestVersion
Review Comment:
nit -- remove commented out code
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #15486:
URL: https://github.com/apache/kafka/pull/15486#issuecomment-2000460307
I was thinking about this change and am wondering if we should bump all the transactional APIs to make it clear this error can be returned. Let me get back to you @sjhajharia -- I will consult some others.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on PR #15486:
URL: https://github.com/apache/kafka/pull/15486#issuecomment-2014388345
Thank you @jolshan for the review.
I have updated the test mentioned. Hopefully it gets into trunk soon.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1523586414
##########
clients/src/main/java/org/apache/kafka/common/protocol/Errors.java:
##########
@@ -392,7 +393,8 @@ public enum Errors {
UNKNOWN_CONTROLLER_ID(116, "This controller ID is not known.", UnknownControllerIdException::new),
UNKNOWN_SUBSCRIPTION_ID(117, "Client sent a push telemetry request with an invalid or outdated subscription ID.", UnknownSubscriptionIdException::new),
TELEMETRY_TOO_LARGE(118, "Client sent a push telemetry request larger than the maximum size the broker will accept.", TelemetryTooLargeException::new),
- INVALID_REGISTRATION(119, "The controller has considered the broker registration to be invalid.", InvalidRegistrationException::new);
+ INVALID_REGISTRATION(119, "The controller has considered the broker registration to be invalid.", InvalidRegistrationException::new),
+ ABORTABLE_TRANSACTION_EXCEPTION(120, "This maps to all errors which can be fixed by aborting the current transaction.", AbortableTransactionException::new);
Review Comment:
nit: let's leave off the "exception" part to be consistent with the other errors here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1525774714
##########
core/src/main/scala/kafka/server/AddPartitionsToTxnManager.scala:
##########
@@ -42,14 +42,26 @@ object AddPartitionsToTxnManager {
val VerificationTimeMsMetricName = "VerificationTimeMs"
}
+/**
+ * This is an enum which handles the Partition Response based on the Produce Request Version and the exact operation
+ * defaultOperation: This is the default workflow which maps to cases when the Produce Request Version was lower than expected or when exercising the offset commit request path
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1525699817
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java:
##########
@@ -1456,7 +1456,7 @@ public <T> CompletableFuture<T> scheduleWriteOperation(
* @param producerEpoch The producer epoch.
* @param timeout The write operation timeout.
* @param op The write operation.
- *
+ * @param apiVersion The Version of the Txn_Offset_Commit request
Review Comment:
can we keep the new line below the last argument?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1522533304
##########
core/src/main/scala/kafka/server/AddPartitionsToTxnManager.scala:
##########
@@ -42,6 +42,11 @@ object AddPartitionsToTxnManager {
val VerificationTimeMsMetricName = "VerificationTimeMs"
}
+object OperationExpected extends Enumeration {
+ type operation = Value
+ val defaultOperation, latestProduceVersion, addPartitionWithoutVerification = Value
Review Comment:
ack
updated the code to use `sealed trait`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1529005304
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -3040,6 +3048,7 @@ class ReplicaManagerTest {
transactionalId = transactionalId,
entriesPerPartition = entriesToAppend,
responseCallback = appendCallback,
+ apiVersionErrorMapper = genericError
Review Comment:
Is there a reason some of these are generic and some are default? I wonder if we can configure the test to use the "highest" enum
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1534705114
##########
core/src/test/scala/unit/kafka/server/AddPartitionsToTxnManagerTest.scala:
##########
@@ -106,27 +106,27 @@ class AddPartitionsToTxnManagerTest {
val transaction2Errors = mutable.Map[TopicPartition, Errors]()
val transaction3Errors = mutable.Map[TopicPartition, Errors]()
- addPartitionsToTxnManager.verifyTransaction(transactionalId1, producerId1, producerEpoch = 0, topicPartitions, setErrors(transaction1Errors))
- addPartitionsToTxnManager.verifyTransaction(transactionalId2, producerId2, producerEpoch = 0, topicPartitions, setErrors(transaction2Errors))
- addPartitionsToTxnManager.verifyTransaction(transactionalId3, producerId3, producerEpoch = 0, topicPartitions, setErrors(transaction3Errors))
+ addPartitionsToTxnManager.verifyTransaction(transactionalId1, producerId1, producerEpoch = 0, topicPartitions, setErrors(transaction1Errors), any())
Review Comment:
for this test, can we use the latest produce version rather than any?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1518062498
##########
clients/src/main/resources/common/message/ProduceRequest.json:
##########
@@ -35,7 +35,10 @@
// Version 9 enables flexible versions.
//
// Version 10 is the same as version 9 (KIP-951).
- "validVersions": "0-10",
+ //
+ // Version 11 is schematically same as Version 10. It will only enable the server to know that
Review Comment:
For now, let's just say Version 11 is the same as 10. We may decide to do a separate bump for the implicit add partitions to transaction.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1518065460
##########
core/src/main/scala/kafka/server/AddPartitionsToTxnManager.scala:
##########
@@ -49,7 +49,8 @@ object AddPartitionsToTxnManager {
*/
class TransactionDataAndCallbacks(val transactionData: AddPartitionsToTxnTransactionCollection,
val callbacks: mutable.Map[String, AddPartitionsToTxnManager.AppendCallback],
- val startTimeMs: mutable.Map[String, Long])
+ val startTimeMs: mutable.Map[String, Long],
+ val produceRequestVersion: Short)
Review Comment:
nice this is what I was thinking π
I noticed that this is a bit tricky for the offset commit path. We will want a given request version there too, but the versions don't map quite the same. I will think a bit on how to address this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1518065460
##########
core/src/main/scala/kafka/server/AddPartitionsToTxnManager.scala:
##########
@@ -49,7 +49,8 @@ object AddPartitionsToTxnManager {
*/
class TransactionDataAndCallbacks(val transactionData: AddPartitionsToTxnTransactionCollection,
val callbacks: mutable.Map[String, AddPartitionsToTxnManager.AppendCallback],
- val startTimeMs: mutable.Map[String, Long])
+ val startTimeMs: mutable.Map[String, Long],
+ val produceRequestVersion: Short)
Review Comment:
nice this is what I was thinking π
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #15486:
URL: https://github.com/apache/kafka/pull/15486#issuecomment-1986154766
I took a quick skim of the tests -- just wanted to confirm we have a test for the old produce request + invalid txn state returned?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1519322254
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java:
##########
@@ -3151,6 +3152,45 @@ public void testInvalidTxnStateIsAnAbortableError() throws Exception {
txnManager.beginTransaction();
}
+
+ @Test
+ public void testAbortableTxnExceptionIsAnAbortableError() throws Exception {
+ ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(123456L, (short) 0);
+ apiVersions.update("0", NodeApiVersions.create(ApiKeys.INIT_PRODUCER_ID.id, (short) 0, (short) 3));
+ TransactionManager txnManager = new TransactionManager(logContext, "textAbortableTxnException", 60000, 100, apiVersions);
+
+ setupWithTransactionState(txnManager);
+ doInitTransactions(txnManager, producerIdAndEpoch);
+
+ txnManager.beginTransaction();
+ txnManager.maybeAddPartition(tp0);
+ client.prepareResponse(buildAddPartitionsToTxnResponseData(0, Collections.singletonMap(tp0, Errors.NONE)));
+ sender.runOnce();
+
+ Future<RecordMetadata> request = appendToAccumulator(tp0);
+ sender.runOnce(); // send request
+ sendIdempotentProducerResponse(0, tp0, Errors.ABORTABLE_TRANSACTION_EXCEPTION, -1);
+
+ // Return InvalidTxnState error. It should be abortable.
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1519327877
##########
core/src/main/scala/kafka/server/AddPartitionsToTxnManager.scala:
##########
@@ -228,11 +232,17 @@ class AddPartitionsToTxnManager(
val tp = new TopicPartition(topicResult.name, partitionResult.partitionIndex)
if (partitionResult.partitionErrorCode != Errors.NONE.code) {
// Producers expect to handle INVALID_PRODUCER_EPOCH in this scenario.
- val code =
+ var code =
if (partitionResult.partitionErrorCode == Errors.PRODUCER_FENCED.code)
Errors.INVALID_PRODUCER_EPOCH.code
else
partitionResult.partitionErrorCode
+ // For backward compatibility with clients.
Review Comment:
agreed. thanks for the suggestion
I will update the code here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1530741929
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -3040,6 +3048,7 @@ class ReplicaManagerTest {
transactionalId = transactionalId,
entriesPerPartition = entriesToAppend,
responseCallback = appendCallback,
+ apiVersionErrorMapper = genericError
Review Comment:
That works for now. Long term, we may want another solution, but we can do in a follow up.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1528996611
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java:
##########
@@ -3523,6 +3524,128 @@ public void testForegroundInvalidStateTransitionIsRecoverable() {
assertFalse(transactionManager.hasOngoingTransaction());
}
+ @Test
+ public void testAbortableTransactionExceptionInInitProducerId() {
+ TransactionalRequestResult initPidResult = transactionManager.initializeTransactions();
+ prepareFindCoordinatorResponse(Errors.NONE, false, CoordinatorType.TRANSACTION, transactionalId);
+ runUntil(() -> transactionManager.coordinator(CoordinatorType.TRANSACTION) != null);
+ assertEquals(brokerNode, transactionManager.coordinator(CoordinatorType.TRANSACTION));
+
+ prepareInitPidResponse(Errors.ABORTABLE_TRANSACTION, false, producerId, RecordBatch.NO_PRODUCER_EPOCH);
+ runUntil(transactionManager::hasError);
+ assertTrue(initPidResult.isCompleted());
+ assertFalse(initPidResult.isSuccessful());
+ assertThrows(AbortableTransactionException.class, initPidResult::await);
+ assertAbortableError(AbortableTransactionException.class);
+ }
+
+ @Test
+ public void testAbortableTransactionExceptionInAddPartitions() {
+ final TopicPartition tp = new TopicPartition("foo", 0);
+
+ doInitTransactions();
+
+ transactionManager.beginTransaction();
+ transactionManager.maybeAddPartition(tp);
+
+ prepareAddPartitionsToTxn(tp, Errors.ABORTABLE_TRANSACTION);
+ runUntil(transactionManager::hasError);
+ assertTrue(transactionManager.lastError() instanceof AbortableTransactionException);
+
+ assertAbortableError(AbortableTransactionException.class);
+ }
+
+ @Test
+ public void testAbortableTransactionExceptionInFindCoordinator() {
+ doInitTransactions();
+
+ transactionManager.beginTransaction();
+ TransactionalRequestResult sendOffsetsResult = transactionManager.sendOffsetsToTransaction(
+ singletonMap(new TopicPartition("foo", 0), new OffsetAndMetadata(39L)), new ConsumerGroupMetadata(consumerGroupId));
+
+ prepareAddOffsetsToTxnResponse(Errors.NONE, consumerGroupId, producerId, epoch);
+ runUntil(() -> !transactionManager.hasPartitionsToAdd());
+
+ prepareFindCoordinatorResponse(Errors.ABORTABLE_TRANSACTION, false, CoordinatorType.GROUP, consumerGroupId);
+ runUntil(transactionManager::hasError);
+ assertTrue(transactionManager.lastError() instanceof AbortableTransactionException);
+
+ runUntil(sendOffsetsResult::isCompleted);
+ assertFalse(sendOffsetsResult.isSuccessful());
+ assertTrue(sendOffsetsResult.error() instanceof AbortableTransactionException);
+
+ assertAbortableError(AbortableTransactionException.class);
+ }
+
+ @Test
+ public void testAbortableTransactionExceptionInEndTxn() throws InterruptedException {
+ doInitTransactions();
+
+ transactionManager.beginTransaction();
+ transactionManager.maybeAddPartition(tp0);
+ TransactionalRequestResult commitResult = transactionManager.beginCommit();
+
+ Future<RecordMetadata> responseFuture = appendToAccumulator(tp0);
+
+ assertFalse(responseFuture.isDone());
+ prepareAddPartitionsToTxnResponse(Errors.NONE, tp0, epoch, producerId);
+ prepareProduceResponse(Errors.NONE, producerId, epoch);
+ prepareEndTxnResponse(Errors.ABORTABLE_TRANSACTION, TransactionResult.COMMIT, producerId, epoch);
+
+ runUntil(commitResult::isCompleted);
+ runUntil(responseFuture::isDone);
+
+ assertThrows(KafkaException.class, commitResult::await);
+ assertFalse(commitResult.isSuccessful());
+ assertTrue(commitResult.isAcked());
+
+ assertAbortableError(AbortableTransactionException.class);
+ // make sure the exception was thrown directly from the follow-up calls.
Review Comment:
nit: was this comment meant to be before further checks? I noticed this test was a little different.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1529794514
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -3040,6 +3048,7 @@ class ReplicaManagerTest {
transactionalId = transactionalId,
entriesPerPartition = entriesToAppend,
responseCallback = appendCallback,
+ apiVersionErrorMapper = genericError
Review Comment:
I will change all of these to use the `genericError` as the supportedOperation
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan merged PR #15486:
URL: https://github.com/apache/kafka/pull/15486
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on PR #15486:
URL: https://github.com/apache/kafka/pull/15486#issuecomment-1987900169
> I took a quick skim of the tests -- just wanted to confirm we have a test for the old produce request + invalid txn state returned?
@jolshan , if you have a look at `core/src/test/scala/unit/kafka/server/AddPartitionsToTxnManagerTest.scala` file, there is a test added around lines 381-390 wherein based on the Produce Request Version, we are testing if the returned error is `Invalid_txn_state` or `AbortableTransactionsException`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1523597267
##########
clients/src/main/java/org/apache/kafka/common/errors/AbortableTransactionException.java:
##########
@@ -0,0 +1,23 @@
+/*
+ * 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.errors;
+
+public class AbortableTransactionException extends ApiException {
Review Comment:
I'm still thinking on this name as well. Since all transactions are technically abortable, and what we want to convey is that the only way forward is to abort, would it make sense to say TransactionAbortableException?
Maybe it doesn't make a huge difference though. π
Wondering if there is a way to make that distinction of abortable vs we need to abort to proceed. I realize though that we are also modeling off of RetriableException here and trying to keep that similar.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1528996611
##########
clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java:
##########
@@ -3523,6 +3524,128 @@ public void testForegroundInvalidStateTransitionIsRecoverable() {
assertFalse(transactionManager.hasOngoingTransaction());
}
+ @Test
+ public void testAbortableTransactionExceptionInInitProducerId() {
+ TransactionalRequestResult initPidResult = transactionManager.initializeTransactions();
+ prepareFindCoordinatorResponse(Errors.NONE, false, CoordinatorType.TRANSACTION, transactionalId);
+ runUntil(() -> transactionManager.coordinator(CoordinatorType.TRANSACTION) != null);
+ assertEquals(brokerNode, transactionManager.coordinator(CoordinatorType.TRANSACTION));
+
+ prepareInitPidResponse(Errors.ABORTABLE_TRANSACTION, false, producerId, RecordBatch.NO_PRODUCER_EPOCH);
+ runUntil(transactionManager::hasError);
+ assertTrue(initPidResult.isCompleted());
+ assertFalse(initPidResult.isSuccessful());
+ assertThrows(AbortableTransactionException.class, initPidResult::await);
+ assertAbortableError(AbortableTransactionException.class);
+ }
+
+ @Test
+ public void testAbortableTransactionExceptionInAddPartitions() {
+ final TopicPartition tp = new TopicPartition("foo", 0);
+
+ doInitTransactions();
+
+ transactionManager.beginTransaction();
+ transactionManager.maybeAddPartition(tp);
+
+ prepareAddPartitionsToTxn(tp, Errors.ABORTABLE_TRANSACTION);
+ runUntil(transactionManager::hasError);
+ assertTrue(transactionManager.lastError() instanceof AbortableTransactionException);
+
+ assertAbortableError(AbortableTransactionException.class);
+ }
+
+ @Test
+ public void testAbortableTransactionExceptionInFindCoordinator() {
+ doInitTransactions();
+
+ transactionManager.beginTransaction();
+ TransactionalRequestResult sendOffsetsResult = transactionManager.sendOffsetsToTransaction(
+ singletonMap(new TopicPartition("foo", 0), new OffsetAndMetadata(39L)), new ConsumerGroupMetadata(consumerGroupId));
+
+ prepareAddOffsetsToTxnResponse(Errors.NONE, consumerGroupId, producerId, epoch);
+ runUntil(() -> !transactionManager.hasPartitionsToAdd());
+
+ prepareFindCoordinatorResponse(Errors.ABORTABLE_TRANSACTION, false, CoordinatorType.GROUP, consumerGroupId);
+ runUntil(transactionManager::hasError);
+ assertTrue(transactionManager.lastError() instanceof AbortableTransactionException);
+
+ runUntil(sendOffsetsResult::isCompleted);
+ assertFalse(sendOffsetsResult.isSuccessful());
+ assertTrue(sendOffsetsResult.error() instanceof AbortableTransactionException);
+
+ assertAbortableError(AbortableTransactionException.class);
+ }
+
+ @Test
+ public void testAbortableTransactionExceptionInEndTxn() throws InterruptedException {
+ doInitTransactions();
+
+ transactionManager.beginTransaction();
+ transactionManager.maybeAddPartition(tp0);
+ TransactionalRequestResult commitResult = transactionManager.beginCommit();
+
+ Future<RecordMetadata> responseFuture = appendToAccumulator(tp0);
+
+ assertFalse(responseFuture.isDone());
+ prepareAddPartitionsToTxnResponse(Errors.NONE, tp0, epoch, producerId);
+ prepareProduceResponse(Errors.NONE, producerId, epoch);
+ prepareEndTxnResponse(Errors.ABORTABLE_TRANSACTION, TransactionResult.COMMIT, producerId, epoch);
+
+ runUntil(commitResult::isCompleted);
+ runUntil(responseFuture::isDone);
+
+ assertThrows(KafkaException.class, commitResult::await);
+ assertFalse(commitResult.isSuccessful());
+ assertTrue(commitResult.isAcked());
+
+ assertAbortableError(AbortableTransactionException.class);
+ // make sure the exception was thrown directly from the follow-up calls.
Review Comment:
nit: was this comment meant to be before further checks?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1518064096
##########
core/src/main/scala/kafka/server/AddPartitionsToTxnManager.scala:
##########
@@ -228,11 +232,17 @@ class AddPartitionsToTxnManager(
val tp = new TopicPartition(topicResult.name, partitionResult.partitionIndex)
if (partitionResult.partitionErrorCode != Errors.NONE.code) {
// Producers expect to handle INVALID_PRODUCER_EPOCH in this scenario.
- val code =
+ var code =
if (partitionResult.partitionErrorCode == Errors.PRODUCER_FENCED.code)
Errors.INVALID_PRODUCER_EPOCH.code
else
partitionResult.partitionErrorCode
+ // For backward compatibility with clients.
Review Comment:
nice π
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1518067902
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1007,7 +1010,8 @@ class ReplicaManager(val config: KafkaConfig,
transactionalId,
producerId,
producerEpoch,
- generalizedCallback
+ generalizedCallback,
+ ApiKeys.PRODUCE.latestVersion
Review Comment:
Hmm this is a bit interesting since this path is usually exercised by the group coordinator. We may want to do something else, or leave a comment here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1525699580
##########
core/src/main/scala/kafka/server/AddPartitionsToTxnManager.scala:
##########
@@ -44,7 +44,7 @@ object AddPartitionsToTxnManager {
/**
* This is an enum which handles the Partition Response based on the Produce Request Version and the exact operation
- * defaultOperation: This is the default workflow which maps to cases when the Produce Request Version was lower than expected or when exercising the offset commit request path
+ * defaultOperation: This is the default workflow which maps to cases when the Produce Request Version or the Txn_offset_commit request was lower than expected
Review Comment:
nit: replace "lower than expected" with "lower than the first version supporting the new error code"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1523717757
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java:
##########
@@ -1504,6 +1507,8 @@ public void handleResponse(AbstractResponse response) {
fatalError(error.exception());
} else if (error == Errors.GROUP_AUTHORIZATION_FAILED) {
abortableError(GroupAuthorizationException.forGroupId(key));
+ } else if (error == Errors.ABORTABLE_TRANSACTION_EXCEPTION) {
Review Comment:
The idea of this error is that we are future proofing. In the future if we want the producer to abort the transaction (say we have a new use case that requires this behavior) we can rely on this error to do the correct thing for older clients. (Or clients 3.8 and later in this case :) )
We ran into this issue a lot when picking error codes for kip-890 part 1. If we had such an error then, we could have used it for the old clients. Instead, we chosen INVALID_TXN_STATE which has inconsistent behavior across clients.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "sjhajharia (via GitHub)" <gi...@apache.org>.
sjhajharia commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1521767609
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1007,7 +1010,8 @@ class ReplicaManager(val config: KafkaConfig,
transactionalId,
producerId,
producerEpoch,
- generalizedCallback
+ generalizedCallback,
+ ApiKeys.PRODUCE.latestVersion
Review Comment:
So for this particular path it would be 0 (default)
And for the new workflow we added, it would be new error (1)
Where can we use the 3rd use case --> that is add partitions rather than verifying?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1521977816
##########
core/src/main/scala/kafka/server/AddPartitionsToTxnManager.scala:
##########
@@ -42,6 +42,11 @@ object AddPartitionsToTxnManager {
val VerificationTimeMsMetricName = "VerificationTimeMs"
}
+object OperationExpected extends Enumeration {
+ type operation = Value
+ val defaultOperation, latestProduceVersion, addPartitionWithoutVerification = Value
Review Comment:
I also think in scala we typically use sealed traits and case objects.
See https://github.com/apache/kafka/blob/58ddd693e69599b177d09c2e384f31e7f5e11171/core/src/main/scala/kafka/controller/KafkaController.scala#L59
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1521780875
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1007,7 +1010,8 @@ class ReplicaManager(val config: KafkaConfig,
transactionalId,
producerId,
producerEpoch,
- generalizedCallback
+ generalizedCallback,
+ ApiKeys.PRODUCE.latestVersion
Review Comment:
This will be done as a follow up and part of KIP-890 part 2
cc: @CalvinConfluent who was also thinking about this part
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16314: Introducing the AbortableTransactionException [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15486:
URL: https://github.com/apache/kafka/pull/15486#discussion_r1521974720
##########
core/src/main/scala/kafka/server/AddPartitionsToTxnManager.scala:
##########
@@ -42,6 +42,11 @@ object AddPartitionsToTxnManager {
val VerificationTimeMsMetricName = "VerificationTimeMs"
}
+object OperationExpected extends Enumeration {
+ type operation = Value
+ val defaultOperation, latestProduceVersion, addPartitionWithoutVerification = Value
Review Comment:
I think we could have something like `default`, `genericError`, and `addPartition`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org