You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "tzulitai (via GitHub)" <gi...@apache.org> on 2023/03/23 22:24:08 UTC
[GitHub] [flink-connector-kafka] tzulitai opened a new pull request, #15: [FLINK-31363] Do not checkpoint a KafkaCommittable if the transaction was empty
tzulitai opened a new pull request, #15:
URL: https://github.com/apache/flink-connector-kafka/pull/15
This PR fixes FLINK-31363 by changing how we handle empty transactions on `prepareCommit`.
Previously, regardless of whether the current transaction is empty or non-empty, we always emit a `KafkaCommittable` for it to be checkpointed by the `CommitterOperator`. The issue: on restore, when we resume the transaction and commit it, we recreate a `FlinkKafkaInternalProducer` that always has the internal `transactionStarted` flag set to `true`, which means that an `EndTxnRequest` will be sent to the brokers for committing the transaction. This results in an `InvalidTxnState` error since on the broker side the transaction hasn't actually been started yet (transactions are lazily started on brokers on the first record sent).
I've considered two possible ways to address this:
1. Store the `transactionStarted` flag in a `KafkaCommittable` alongside other txn metadata. Then, on restore, on the recreated producer, we set the internal `transactionStarted` accordingly to what the checkpoint says.
2. Never checkpoint a `KafkaCommitable` if the transaction is empty. In this case, any `KafkaCommittable` restored from a checkpoint always has some data in them, and therefore it is correct to always set the internal `transactionStarted` flag to `true` on the recreated producer.
This PR chooses to go with approach 2.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-kafka] boring-cyborg[bot] commented on pull request #15: [FLINK-31363] Do not checkpoint a KafkaCommittable if the transaction was empty
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #15:
URL: https://github.com/apache/flink-connector-kafka/pull/15#issuecomment-1481994152
Thanks for opening this pull request! Please check out our contributing guidelines. (https://flink.apache.org/contributing/how-to-contribute.html)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-kafka] RamanVerma commented on a diff in pull request #15: [FLINK-31363] Do not checkpoint a KafkaCommittable if the transaction was empty
Posted by "RamanVerma (via GitHub)" <gi...@apache.org>.
RamanVerma commented on code in PR #15:
URL: https://github.com/apache/flink-connector-kafka/pull/15#discussion_r1151154666
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducer.java:
##########
@@ -67,6 +72,14 @@ private static Properties withTransactionalId(
return props;
}
+ @Override
+ public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
+ if (inTransaction) {
+ hasRecordsInTransaction = true;
Review Comment:
Should this boolean be set in the callback, in the successful send scenario?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-kafka] tzulitai commented on a diff in pull request #15: [FLINK-31363] Do not checkpoint a KafkaCommittable if the transaction was empty
Posted by "tzulitai (via GitHub)" <gi...@apache.org>.
tzulitai commented on code in PR #15:
URL: https://github.com/apache/flink-connector-kafka/pull/15#discussion_r1151177809
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducer.java:
##########
@@ -67,6 +72,14 @@ private static Properties withTransactionalId(
return props;
}
+ @Override
+ public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
+ if (inTransaction) {
+ hasRecordsInTransaction = true;
Review Comment:
hmm, that's a good point. I think the question to ask is: is it incorrect to set this flag (to allow a `KafkaCommittable` to be generated for the txn at pre-commit time) preemptively, instead of only setting it when data has actually been successfully written?
I think the answer is that it is not incorrect, so it is ok to leave this as is. Reasoning is as follows:
- At pre-commit time and performing flush, if some data failed to be flushed, the pre-commit will fail so a `KafkaCommittable` will not be checkpointed for the txn anyways. In this scenario, the `hasRecordsInTransaction` flag is irrelevant no matter its value.
- If all records are correctly flushed, then good; a `KafkaCommittable` should be generated for the txn. We're good here because we've alraedy preemptively set the `hasRecordsInTransaction` flag.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-kafka] tzulitai commented on a diff in pull request #15: [FLINK-31363] Do not checkpoint a KafkaCommittable if the transaction was empty
Posted by "tzulitai (via GitHub)" <gi...@apache.org>.
tzulitai commented on code in PR #15:
URL: https://github.com/apache/flink-connector-kafka/pull/15#discussion_r1151180651
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducer.java:
##########
@@ -67,6 +72,14 @@ private static Properties withTransactionalId(
return props;
}
+ @Override
+ public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
+ if (inTransaction) {
+ hasRecordsInTransaction = true;
Review Comment:
Speaking of which, it might give a more complete picture of this interaction after I rebase this PR branch on top of the latest changes (to include the fix that adds `checkAsyncExceptions`).
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-kafka] RamanVerma commented on a diff in pull request #15: [FLINK-31363] Do not checkpoint a KafkaCommittable if the transaction was empty
Posted by "RamanVerma (via GitHub)" <gi...@apache.org>.
RamanVerma commented on code in PR #15:
URL: https://github.com/apache/flink-connector-kafka/pull/15#discussion_r1151169873
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducerITCase.java:
##########
@@ -81,12 +80,58 @@ void testInitTransactionId() {
reuse.flush();
reuse.abortTransaction();
}
- assertNumTransactions(i);
+ assertNumTransactions(i, transactionIdPrefix);
assertThat(readRecords(topic).count()).isEqualTo(i / 2);
}
}
}
+ @Test
+ void testCommitResumedTransaction() {
+ final String topic = "test-commit-resumed-transaction";
+ final String transactionIdPrefix = "testCommitResumedTransaction-";
+ final String transactionalId = transactionIdPrefix + "id";
+
+ KafkaCommittable snapshottedCommittable;
+ try (FlinkKafkaInternalProducer<String, String> producer =
+ new FlinkKafkaInternalProducer<>(getProperties(), transactionalId)) {
+ producer.initTransactions();
+ producer.beginTransaction();
+ producer.send(new ProducerRecord<>(topic, "test-value"));
+ producer.flush();
+ snapshottedCommittable = KafkaCommittable.of(producer, ignored -> {});
+ }
+
+ try (FlinkKafkaInternalProducer<String, String> resumedProducer =
+ new FlinkKafkaInternalProducer<>(getProperties(), transactionalId)) {
+ resumedProducer.resumeTransaction(
+ snapshottedCommittable.getProducerId(), snapshottedCommittable.getEpoch());
+ resumedProducer.commitTransaction();
+ }
+
+ assertNumTransactions(1, transactionIdPrefix);
+ assertThat(readRecords(topic).count()).isEqualTo(1);
+ }
+
+ @Test
+ void testCommitResumedEmptyTransactionShouldFail() {
+ KafkaCommittable snapshottedCommittable;
+ try (FlinkKafkaInternalProducer<String, String> producer =
+ new FlinkKafkaInternalProducer<>(getProperties(), "dummy")) {
+ producer.initTransactions();
+ producer.beginTransaction();
+ snapshottedCommittable = KafkaCommittable.of(producer, ignored -> {});
+ }
+
+ try (FlinkKafkaInternalProducer<String, String> resumedProducer =
+ new FlinkKafkaInternalProducer<>(getProperties(), "dummy")) {
+ resumedProducer.resumeTransaction(
+ snapshottedCommittable.getProducerId(), snapshottedCommittable.getEpoch());
+
+ assertThatThrownBy(resumedProducer::commitTransaction);
Review Comment:
nit: Can we check the exception type also using
`isInstanceOf(<exception type>.class)`
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-kafka] tzulitai commented on a diff in pull request #15: [FLINK-31363] Do not checkpoint a KafkaCommittable if the transaction was empty
Posted by "tzulitai (via GitHub)" <gi...@apache.org>.
tzulitai commented on code in PR #15:
URL: https://github.com/apache/flink-connector-kafka/pull/15#discussion_r1151177809
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducer.java:
##########
@@ -67,6 +72,14 @@ private static Properties withTransactionalId(
return props;
}
+ @Override
+ public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
+ if (inTransaction) {
+ hasRecordsInTransaction = true;
Review Comment:
hmm, that's a good point. I think the question to ask is: is it incorrect to set this flag (to allow a `KafkaCommittable` to be generated for the txn at pre-commit time) preemptively, instead of only setting it when data has actually been written?
I think the answer is that it is not incorrect, so it is ok to leave this as is. Reasoning is as follows:
- At pre-commit time and performing flush, if some data failed to be flushed, the pre-commit will fail so a `KafkaCommittable` will not be checkpointed for the txn anyways. In this scenario, the `hasRecordsInTransaction` flag is irrelevant anyways.
- If all records are correctly flushed, then good; a `KafkaCommittable` should be generated for the txn. We're good here because we've alraedy preemptively set the `hasRecordsInTransaction` flag.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-kafka] tzulitai closed pull request #15: [FLINK-31363] Do not checkpoint a KafkaCommittable if the transaction was empty
Posted by "tzulitai (via GitHub)" <gi...@apache.org>.
tzulitai closed pull request #15: [FLINK-31363] Do not checkpoint a KafkaCommittable if the transaction was empty
URL: https://github.com/apache/flink-connector-kafka/pull/15
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-kafka] RamanVerma commented on a diff in pull request #15: [FLINK-31363] Do not checkpoint a KafkaCommittable if the transaction was empty
Posted by "RamanVerma (via GitHub)" <gi...@apache.org>.
RamanVerma commented on code in PR #15:
URL: https://github.com/apache/flink-connector-kafka/pull/15#discussion_r1149828787
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducerITCase.java:
##########
@@ -81,12 +81,64 @@ void testInitTransactionId() {
reuse.flush();
reuse.abortTransaction();
}
- assertNumTransactions(i);
+ assertNumTransactions(i, transactionIdPrefix);
assertThat(readRecords(topic).count()).isEqualTo(i / 2);
}
}
}
+ @Test
+ void testCommitResumedTransaction() {
+ final String topic = "test-commit-resumed-transaction";
+ final String transactionIdPrefix = "testCommitResumedTransaction-";
+ final String transactionalId = transactionIdPrefix + "id";
+
+ KafkaCommittable snapshottedCommittable;
+ try (FlinkKafkaInternalProducer<String, String> producer =
+ new FlinkKafkaInternalProducer<>(getProperties(), transactionalId)) {
+ producer.initTransactions();
+ producer.beginTransaction();
+ producer.send(new ProducerRecord<>(topic, "test-value"));
+ producer.flush();
+ snapshottedCommittable = KafkaCommittable.of(producer, ignored -> {});
+ }
+
+ try (FlinkKafkaInternalProducer<String, String> resumedProducer =
+ new FlinkKafkaInternalProducer<>(getProperties(), transactionalId)) {
+ resumedProducer.resumeTransaction(
+ snapshottedCommittable.getProducerId(), snapshottedCommittable.getEpoch());
+ resumedProducer.commitTransaction();
+ }
+
+ assertNumTransactions(1, transactionIdPrefix);
+ assertThat(readRecords(topic).count()).isEqualTo(1);
+ }
+
+ @Test
+ void testCommitResumedEmptyTransactionShouldFail() {
+ KafkaCommittable snapshottedCommittable;
+ try (FlinkKafkaInternalProducer<String, String> producer =
+ new FlinkKafkaInternalProducer<>(getProperties(), "dummy")) {
+ producer.initTransactions();
+ producer.beginTransaction();
+ snapshottedCommittable = KafkaCommittable.of(producer, ignored -> {});
+ }
+
+ try (FlinkKafkaInternalProducer<String, String> resumedProducer =
+ new FlinkKafkaInternalProducer<>(getProperties(), "dummy")) {
+ resumedProducer.resumeTransaction(
+ snapshottedCommittable.getProducerId(), snapshottedCommittable.getEpoch());
+
+ try {
+ resumedProducer.commitTransaction();
Review Comment:
You can use `assertThatThrownBy` 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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-kafka] tzulitai commented on a diff in pull request #15: [FLINK-31363] Do not checkpoint a KafkaCommittable if the transaction was empty
Posted by "tzulitai (via GitHub)" <gi...@apache.org>.
tzulitai commented on code in PR #15:
URL: https://github.com/apache/flink-connector-kafka/pull/15#discussion_r1151177809
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducer.java:
##########
@@ -67,6 +72,14 @@ private static Properties withTransactionalId(
return props;
}
+ @Override
+ public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
+ if (inTransaction) {
+ hasRecordsInTransaction = true;
Review Comment:
hmm, that's a good point. I think the question to ask is: is it incorrect to set this flag (to allow a `KafkaCommittable` to be generated for the txn at pre-commit time) preemptively, instead of only setting it when data has actually been successfully written?
I think the answer is that it is not incorrect, so it is ok to leave this as is. Reasoning is as follows:
- At pre-commit time and performing flush, if some data failed to be flushed, the pre-commit will fail so a `KafkaCommittable` will not be checkpointed for the txn anyways. In this scenario, the `hasRecordsInTransaction` flag is irrelevant anyways.
- If all records are correctly flushed, then good; a `KafkaCommittable` should be generated for the txn. We're good here because we've alraedy preemptively set the `hasRecordsInTransaction` flag.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-kafka] tzulitai commented on pull request #15: [FLINK-31363] Do not checkpoint a KafkaCommittable if the transaction was empty
Posted by "tzulitai (via GitHub)" <gi...@apache.org>.
tzulitai commented on PR #15:
URL: https://github.com/apache/flink-connector-kafka/pull/15#issuecomment-1481995033
cc @RamanVerma @Gerrrr for review
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-connector-kafka] tzulitai commented on pull request #15: [FLINK-31363] Do not checkpoint a KafkaCommittable if the transaction was empty
Posted by "tzulitai (via GitHub)" <gi...@apache.org>.
tzulitai commented on PR #15:
URL: https://github.com/apache/flink-connector-kafka/pull/15#issuecomment-1489305429
thanks for the review @RamanVerma, merging this now!
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org