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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
