You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "AHeise (via GitHub)" <gi...@apache.org> on 2025/04/08 07:48:11 UTC
[PR] [FLINK-37622] Fix KafkaSink in BATCH [flink-connector-kafka]
AHeise opened a new pull request, #166:
URL: https://github.com/apache/flink-connector-kafka/pull/166
Since BATCH doesn't call snapshotState(), we never rotate the producer in batch mode. Hence, on closing the producer in the write failover region, we also abort the ongoing transaction. The committer in the next region is then failing.
This commit generalizes the client side transaction state machine and adds a new state for precommitted. Then the writer must abort only those transactions that are not precommitted.
--
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
Re: [PR] [FLINK-37622] Fix KafkaSink in BATCH [flink-connector-kafka]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise merged PR #166:
URL: https://github.com/apache/flink-connector-kafka/pull/166
--
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
Re: [PR] [FLINK-37622] Fix KafkaSink in BATCH [flink-connector-kafka]
Posted by "fapaul (via GitHub)" <gi...@apache.org>.
fapaul commented on code in PR #166:
URL: https://github.com/apache/flink-connector-kafka/pull/166#discussion_r2032772897
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/FlinkKafkaInternalProducer.java:
##########
@@ -97,33 +96,39 @@ public void flush() {
public void beginTransaction() throws ProducerFencedException {
super.beginTransaction();
LOG.debug("beginTransaction {}", transactionalId);
- inTransaction = true;
+ transactionState = TransactionState.IN_TRANSACTION;
}
@Override
public void abortTransaction() throws ProducerFencedException {
LOG.debug("abortTransaction {}", transactionalId);
- checkState(inTransaction, "Transaction was not started");
- inTransaction = false;
- hasRecordsInTransaction = false;
+ checkState(isInTransaction(), "Transaction was not started");
+ transactionState = TransactionState.NOT_IN_TRANSACTION;
super.abortTransaction();
}
@Override
public void commitTransaction() throws ProducerFencedException {
LOG.debug("commitTransaction {}", transactionalId);
- checkState(inTransaction, "Transaction was not started");
- inTransaction = false;
- hasRecordsInTransaction = false;
+ checkState(isInTransaction(), "Transaction was not started");
+ transactionState = TransactionState.NOT_IN_TRANSACTION;
super.commitTransaction();
}
public boolean isInTransaction() {
- return inTransaction;
+ return transactionState != TransactionState.NOT_IN_TRANSACTION;
}
public boolean hasRecordsInTransaction() {
- return hasRecordsInTransaction;
+ return TransactionState.DATA_IN_TRANSACTION.compareTo(transactionState) <= 0;
Review Comment:
Nit: I do not really like relying on the order inside the enum. We do not have a unit test to verify that and it's easy to break.
Can't we use
```java
TransactionState.DATA_IN_TRANSACTION == transactionState
```
--
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