slachiewicz commented on code in PR #314:
URL:
https://github.com/apache/flink-connector-kafka/pull/314#discussion_r4054035209
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/TransactionAbortStrategyContextImpl.java:
##########
@@ -111,7 +136,50 @@ public Set<String> getPrefixesToAbort() {
@Override
public Set<String> getPrecommittedTransactionalIds() {
- return precommittedTransactionIds;
+ return Collections.unmodifiableSet(precommittedTransactions.keySet());
+ }
+
+ @Override
+ public boolean isPrecommittedTransactionSuperseded(String transactionalId)
{
Review Comment:
Yes. `TransactionAbortStrategyImplTest` drives the strategy through a fake
context, so this method itself runs only in `ExactlyOnceKafkaWriterITCase` and
`KafkaSinkRecoveryITCase`, because the decision needs
`Admin.describeTransactions`. Those cover the newer-epoch case; the "state has
no epoch" branch is covered through `KafkaWriterStateSerializerTest` reading v2
bytes as unknown. If you want the full matrix (id unknown to the broker, equal
epoch, different producer id) without a broker, I can add a unit test with a
stub `Admin`.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/internal/TransactionAbortStrategyContextImpl.java:
##########
@@ -111,7 +136,50 @@ public Set<String> getPrefixesToAbort() {
@Override
public Set<String> getPrecommittedTransactionalIds() {
- return precommittedTransactionIds;
+ return Collections.unmodifiableSet(precommittedTransactions.keySet());
+ }
+
+ @Override
+ public boolean isPrecommittedTransactionSuperseded(String transactionalId)
{
+ CheckpointTransaction transaction =
precommittedTransactions.get(transactionalId);
+ if (transaction == null || !transaction.hasKnownEpoch()) {
+ // state written before v3 did not record the epoch; keep the old
behavior
+ return false;
+ }
+ TransactionDescription description =
describePrecommitted().get(transactionalId);
+ if (description == null) {
+ // the broker does not know the id any more; nothing to abort
+ return false;
+ }
+ boolean superseded =
+ description.producerId() != transaction.getProducerId()
+ || description.producerEpoch() >
transaction.getEpoch();
+ if (superseded) {
+ LOG.info(
+ "Recovered transaction {} was opened with producer id {}
and epoch {}, but the broker now holds producer id {} and epoch {} in state {}",
+ transactionalId,
+ transaction.getProducerId(),
+ transaction.getEpoch(),
+ description.producerId(),
+ description.producerEpoch(),
+ description.state());
+ }
+ return superseded;
+ }
+
+ @Override
+ public void abandonPrecommittedTransaction(String transactionalId) {
+ precommittedTransactions.remove(transactionalId);
+ precommittedTransactionAbandoner.accept(transactionalId);
+ }
+
+ private Map<String, TransactionDescription> describePrecommitted() {
+ if (precommittedDescriptions == null) {
+ precommittedDescriptions =
+ AdminUtils.describeTransactions(
Review Comment:
Once per writer initialization, so once per subtask start or restart:
`abortLingeringTransactions` creates the context in `initialize()`, the first
superseded check issues one batched `describeTransactions` for all precommitted
ids, and the result is cached for the lifetime of the context, which ends with
that call. Nothing on the checkpoint or record path.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriterStateSerializer.java:
##########
@@ -31,11 +32,17 @@
import static org.apache.flink.connector.kafka.sink.KafkaWriterState.UNKNOWN;
-/** A serializer used to serialize {@link KafkaWriterState}. */
+/**
+ * A serializer used to serialize {@link KafkaWriterState}.
+ *
+ * <p>Version 3 adds the producer id and epoch of every precommitted
transaction, so that a recovery
Review Comment:
Forward-only, like the 1 to 2 bump: a connector without this change throws
`IOException("Unknown version: 3")` on restore, so a rollback needs a savepoint
taken by the older version. Backports to v5.0 and v4.0 keep the version
consistent across the supported lines.
--
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]
