You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ar...@apache.org on 2021/09/01 06:29:32 UTC
[flink] 10/11: [FLINK-23854][connectors/kafka] Reliably abort
lingering transactions in Kafka.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 96a3f123a1713b63c714aae7ac0032a2463638e9
Author: Arvid Heise <ar...@ververica.com>
AuthorDate: Fri Aug 27 11:13:08 2021 +0200
[FLINK-23854][connectors/kafka] Reliably abort lingering transactions in Kafka.
The new approach depends on successively aborting transactions until an unused transaction is ensured. This approach covers all edge cases including failures before first checkpoint and downscaling without checkpoints.
---
flink-connectors/flink-connector-kafka/pom.xml | 6 +
.../kafka/sink/FlinkKafkaInternalProducer.java | 20 ++
.../connector/kafka/sink/KafkaTransactionLog.java | 276 ---------------------
.../flink/connector/kafka/sink/KafkaWriter.java | 166 +++++--------
.../connector/kafka/sink/KafkaWriterState.java | 23 +-
.../kafka/sink/KafkaWriterStateSerializer.java | 6 +-
.../connector/kafka/sink/TransactionAborter.java | 131 ++++++++++
.../kafka/sink/TransactionalIdFactory.java | 42 ----
.../sink/FlinkKafkaInternalProducerITCase.java | 122 +++++++++
.../connector/kafka/sink/KafkaTransactionLog.java | 247 ++++++++++++++++++
.../kafka/sink/KafkaTransactionLogITCase.java | 136 ++++------
.../kafka/sink/KafkaWriterStateSerializerTest.java | 2 +-
.../kafka/sink/TransactionIdFactoryTest.java | 40 ---
13 files changed, 644 insertions(+), 573 deletions(-)
diff --git a/flink-connectors/flink-connector-kafka/pom.xml b/flink-connectors/flink-connector-kafka/pom.xml
index 0da6052..009a184 100644
--- a/flink-connectors/flink-connector-kafka/pom.xml
+++ b/flink-connectors/flink-connector-kafka/pom.xml
@@ -125,6 +125,12 @@ under the License.
</dependency>
<dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>junit-jupiter</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-tests</artifactId>
<version>${project.version}</version>
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducer.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducer.java
index dadee58..c04f73e 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducer.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducer.java
@@ -33,6 +33,7 @@ import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.time.Duration;
import java.util.Properties;
+import java.util.concurrent.TimeUnit;
import static org.apache.flink.util.Preconditions.checkState;
@@ -49,6 +50,7 @@ class FlinkKafkaInternalProducer<K, V> extends KafkaProducer<K, V> {
@Nullable private String transactionalId;
private volatile boolean inTransaction;
+ private volatile boolean closed;
public FlinkKafkaInternalProducer(Properties properties, @Nullable String transactionalId) {
super(withTransactionalId(properties, transactionalId));
@@ -98,9 +100,27 @@ class FlinkKafkaInternalProducer<K, V> extends KafkaProducer<K, V> {
@Override
public void close() {
+ closed = true;
+ flush();
super.close(Duration.ZERO);
}
+ @Override
+ public void close(Duration timeout) {
+ closed = true;
+ super.close(timeout);
+ }
+
+ @Override
+ public void close(long timeout, TimeUnit unit) {
+ closed = true;
+ super.close(timeout, unit);
+ }
+
+ public boolean isClosed() {
+ return closed;
+ }
+
@Nullable
public String getTransactionalId() {
return transactionalId;
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaTransactionLog.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaTransactionLog.java
deleted file mode 100644
index 61370c3..0000000
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaTransactionLog.java
+++ /dev/null
@@ -1,276 +0,0 @@
-/*
- * 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.flink.connector.kafka.sink;
-
-import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList;
-import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableSet;
-
-import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.consumer.ConsumerRecords;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
-import org.apache.kafka.common.KafkaException;
-import org.apache.kafka.common.PartitionInfo;
-import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.serialization.ByteArrayDeserializer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.nio.ByteBuffer;
-import java.nio.charset.StandardCharsets;
-import java.time.Duration;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-import java.util.ListIterator;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Properties;
-import java.util.Set;
-import java.util.UUID;
-import java.util.function.Consumer;
-import java.util.stream.Collectors;
-
-import static org.apache.flink.connector.kafka.sink.KafkaTransactionLog.TransactionState.CompleteAbort;
-import static org.apache.flink.connector.kafka.sink.KafkaTransactionLog.TransactionState.CompleteCommit;
-import static org.apache.flink.connector.kafka.sink.KafkaTransactionLog.TransactionState.Dead;
-import static org.apache.flink.util.Preconditions.checkNotNull;
-import static org.apache.kafka.common.internals.Topic.TRANSACTION_STATE_TOPIC_NAME;
-
-/**
- * This class is responsible to provide the format of the used transationalIds and in case of an
- * application restart query the open transactions and decide which must be aborted.
- */
-class KafkaTransactionLog implements AutoCloseable {
-
- private static final Logger LOG = LoggerFactory.getLogger(KafkaTransactionLog.class);
- private static final Duration CONSUMER_POLL_DURATION = Duration.ofSeconds(1);
- private static final Set<TransactionState> TERMINAL_TRANSACTION_STATES =
- ImmutableSet.of(CompleteCommit, CompleteAbort, Dead);
- private static final int SUPPORTED_KAFKA_SCHEMA_VERSION = 0;
-
- private final KafkaConsumer<byte[], byte[]> consumer;
- private final KafkaWriterState main;
- private final TransactionsToAbortChecker transactionToAbortChecker;
-
- /**
- * Constructor creating a KafkaTransactionLog.
- *
- * @param kafkaConfig used to configure the {@link KafkaConsumer} to query the topic containing
- * the transaction information
- * @param main the {@link KafkaWriterState} which was previously snapshotted by this subtask
- * @param others the {@link KafkaWriterState}s which are from different subtasks i.e. in case of
- * a scale-in
- * @param numberOfParallelSubtasks current number of parallel sink tasks
- */
- KafkaTransactionLog(
- Properties kafkaConfig,
- KafkaWriterState main,
- List<KafkaWriterState> others,
- int numberOfParallelSubtasks) {
- this.main = checkNotNull(main, "mainState");
- checkNotNull(others, "othersState");
- final Map<Integer, Long> subtaskIdCheckpointOffsetMapping =
- new ImmutableList.Builder<KafkaWriterState>()
- .add(main).addAll(others).build().stream()
- .collect(
- Collectors.toMap(
- KafkaWriterState::getSubtaskId,
- KafkaWriterState::getTransactionalIdOffset));
- final Properties consumerConfig = new Properties();
- consumerConfig.putAll(checkNotNull(kafkaConfig, "kafkaConfig"));
- consumerConfig.put("enable.auto.commit", false);
- consumerConfig.put("key.deserializer", ByteArrayDeserializer.class.getName());
- consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString());
- consumerConfig.put("auto.offset.reset", "earliest");
- consumerConfig.put("value.deserializer", ByteArrayDeserializer.class.getName());
- consumerConfig.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
- this.transactionToAbortChecker =
- new TransactionsToAbortChecker(
- numberOfParallelSubtasks,
- subtaskIdCheckpointOffsetMapping,
- main.getSubtaskId());
- this.consumer = new KafkaConsumer<>(consumerConfig);
- this.consumer.assign(getAllPartitions());
- }
-
- /**
- * This method queries Kafka's internal transaction topic and filters the transactions for the
- * following rules.
- * <li>transaction is in no terminal state {@link
- * KafkaTransactionLog#TERMINAL_TRANSACTION_STATES}
- * <li>transactionalIdPrefix equals the one from {@link #main}
- * <li>Applies the rules from {@link TransactionsToAbortChecker}
- *
- * @return all transactionIds which must be aborted before starting new transactions.
- * @throws KafkaException if no transaction information can be fetched. (e.g. no access rights
- * to __transaction_state topic)
- */
- public List<String> getTransactionsToAbort() throws KafkaException {
- final Map<Integer, Map<Long, String>> openTransactions = new HashMap<>();
- final List<TopicPartition> partitions = getAllPartitions();
- final Map<TopicPartition, Long> endOffsets = consumer.endOffsets(partitions);
- do {
- // We have to call poll() first before we have any partitions assigned
- ConsumerRecords<byte[], byte[]> records = consumer.poll(CONSUMER_POLL_DURATION);
- records.records(TRANSACTION_STATE_TOPIC_NAME)
- .forEach(maybeAddTransaction(openTransactions));
- } while (!hasReadAllRecords(endOffsets, partitions));
-
- return transactionToAbortChecker.getTransactionsToAbort(openTransactions);
- }
-
- private boolean hasReadAllRecords(
- Map<TopicPartition, Long> endOffsets, List<TopicPartition> partitions) {
- final ListIterator<TopicPartition> it = partitions.listIterator();
- while (it.hasNext()) {
- final TopicPartition partition = it.next();
- final long endOffset = endOffsets.get(partition);
- if (endOffset == 0 || consumer.position(partition) >= endOffset) {
- // Remove finished partitions to decrease the look-up space
- it.remove();
- continue;
- }
- return false;
- }
- return true;
- }
-
- private List<TopicPartition> getAllPartitions() {
- final List<PartitionInfo> partitionInfos =
- consumer.partitionsFor(TRANSACTION_STATE_TOPIC_NAME);
- return partitionInfos.stream()
- .filter(info -> info.topic().equals(TRANSACTION_STATE_TOPIC_NAME))
- .map(info -> new TopicPartition(info.topic(), info.partition()))
- .collect(Collectors.toList());
- }
-
- private Consumer<ConsumerRecord<byte[], byte[]>> maybeAddTransaction(
- Map<Integer, Map<Long, String>> openTransactions) {
- return record -> {
- final ByteBuffer keyBuffer = ByteBuffer.wrap(record.key());
- checkKafkaSchemaVersionMatches(keyBuffer);
- // Ignore 2 bytes because Kafka's internal representation
- keyBuffer.getShort();
- final String transactionalId = StandardCharsets.UTF_8.decode(keyBuffer).toString();
-
- final Optional<KafkaWriterState> openTransactionOpt =
- TransactionalIdFactory.parseKafkaWriterState(transactionalId);
- // If the transactionalId does not follow the format ignore it
- if (!openTransactionOpt.isPresent()) {
- return;
- }
-
- final KafkaWriterState openTransaction = openTransactionOpt.get();
- // If the transactionalId prefixes differ ignore
- if (!openTransaction
- .getTransactionalIdPrefix()
- .equals(main.getTransactionalIdPrefix())) {
- LOG.debug(
- "The transactionalId prefixes differ. Open: {}, Recovered: {}",
- openTransaction.getTransactionalIdPrefix(),
- main.getTransactionalIdPrefix());
- return;
- }
-
- final ByteBuffer valueBuffer = ByteBuffer.wrap(record.value());
- checkKafkaSchemaVersionMatches(valueBuffer);
- final TransactionState state =
- TransactionState.fromByte(readTransactionState(valueBuffer));
-
- LOG.debug("Transaction {} is in state {}", transactionalId, state);
-
- final int openSubtaskIndex = openTransaction.getSubtaskId();
- final long openCheckpointOffset = openTransaction.getTransactionalIdOffset();
-
- // If the transaction is in a final state ignore it
- if (isTransactionInFinalState(state)) {
- openTransactions.get(openSubtaskIndex).remove(openCheckpointOffset);
- return;
- }
-
- if (openTransactions.containsKey(openSubtaskIndex)) {
- openTransactions.get(openSubtaskIndex).put(openCheckpointOffset, transactionalId);
- } else {
- final Map<Long, String> map = new HashMap<>();
- map.put(openCheckpointOffset, transactionalId);
- openTransactions.put(openSubtaskIndex, map);
- }
- };
- }
-
- @Override
- public void close() {
- consumer.close();
- }
-
- private static boolean isTransactionInFinalState(TransactionState state) {
- return TERMINAL_TRANSACTION_STATES.contains(state);
- }
-
- private static byte readTransactionState(ByteBuffer buffer) {
- // producerId
- buffer.getLong();
- // epoch
- buffer.getShort();
- // transactionTimeout
- buffer.getInt();
- // statusKey
- return buffer.get();
- }
-
- enum TransactionState {
- Empty(Byte.parseByte("0")),
- Ongoing(Byte.parseByte("1")),
- PrepareCommit(Byte.parseByte("2")),
- PrepareAbort(Byte.parseByte("3")),
- CompleteCommit(Byte.parseByte("4")),
- CompleteAbort(Byte.parseByte("5")),
- Dead(Byte.parseByte("6")),
- PrepareEpochFence(Byte.parseByte("7"));
-
- private static final Map<Byte, TransactionState> BYTE_TO_STATE =
- Arrays.stream(TransactionState.values())
- .collect(Collectors.toMap(e -> e.state, e -> e));
-
- private final byte state;
-
- TransactionState(byte state) {
- this.state = state;
- }
-
- static TransactionState fromByte(byte state) {
- final TransactionState transactionState = BYTE_TO_STATE.get(state);
- if (transactionState == null) {
- throw new IllegalArgumentException(
- String.format("The given state %s is not supported.", state));
- }
- return transactionState;
- }
- }
-
- private static void checkKafkaSchemaVersionMatches(ByteBuffer buffer) {
- final short version = buffer.getShort();
- if (version != SUPPORTED_KAFKA_SCHEMA_VERSION) {
- throw new IllegalStateException(
- String.format(
- "Kafka has changed the schema version from %s to %s",
- SUPPORTED_KAFKA_SCHEMA_VERSION, version));
- }
- }
-}
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
index ae2c922..02af7ff 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
@@ -29,12 +29,12 @@ import org.apache.flink.streaming.connectors.kafka.internals.metrics.KafkaMetric
import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList;
+import org.apache.flink.shaded.guava30.com.google.common.collect.Lists;
import org.apache.flink.shaded.guava30.com.google.common.io.Closer;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.Metric;
import org.apache.kafka.common.MetricName;
import org.slf4j.Logger;
@@ -43,8 +43,6 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
import java.io.IOException;
-import java.time.Duration;
-import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -52,8 +50,6 @@ import java.util.Map;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
-import java.util.function.Function;
-import java.util.stream.Collectors;
import static org.apache.flink.util.Preconditions.checkNotNull;
import static org.apache.flink.util.Preconditions.checkState;
@@ -91,6 +87,7 @@ class KafkaWriter<IN> implements SinkWriter<IN, KafkaCommittable, KafkaWriterSta
private KafkaWriterState kafkaWriterState;
private final Closer closer = Closer.create();
@Nullable private volatile Exception producerAsyncException;
+ private long lastCheckpointId;
private boolean closed = false;
private long lastSync = System.currentTimeMillis();
@@ -150,9 +147,11 @@ class KafkaWriter<IN> implements SinkWriter<IN, KafkaCommittable, KafkaWriterSta
} catch (Exception e) {
throw new FlinkRuntimeException("Cannot initialize schema.", e);
}
- this.kafkaWriterState =
- recoverAndInitializeState(checkNotNull(recoveredStates, "recoveredStates"));
- this.currentProducer = createProducer();
+ lastCheckpointId = sinkInitContext.getRestoredCheckpointId().orElse(0);
+ abortLingeringTransactions(
+ checkNotNull(recoveredStates, "recoveredStates"), lastCheckpointId + 1);
+ this.kafkaWriterState = new KafkaWriterState(transactionalIdPrefix);
+ this.currentProducer = createProducer(lastCheckpointId + 1);
registerMetricSync();
}
@@ -168,80 +167,49 @@ class KafkaWriter<IN> implements SinkWriter<IN, KafkaCommittable, KafkaWriterSta
@Override
public List<KafkaCommittable> prepareCommit(boolean flush) {
flushRecords(flush);
- List<KafkaCommittable> committables = precommit();
- currentProducer = createProducer();
- return committables;
+ return precommit();
}
@Override
public List<KafkaWriterState> snapshotState(long checkpointId) throws IOException {
+ currentProducer = createProducer(checkpointId + 1);
return ImmutableList.of(kafkaWriterState);
}
@Override
public void close() throws Exception {
+
if (currentProducer.isInTransaction()) {
currentProducer.abortTransaction();
}
- currentProducer.flush();
closed = true;
closer.close();
+ checkState(currentProducer.isClosed());
}
- private KafkaWriterState recoverAndInitializeState(List<KafkaWriterState> recoveredStates) {
- final int subtaskId = kafkaSinkContext.getParallelInstanceId();
- if (recoveredStates.isEmpty()) {
- final KafkaWriterState state =
- new KafkaWriterState(transactionalIdPrefix, subtaskId, 0);
- abortTransactions(getTransactionsToAbort(state, new ArrayList<>()));
- return state;
- }
- final Map<Integer, KafkaWriterState> taskOffsetMapping =
- recoveredStates.stream()
- .collect(
- Collectors.toMap(
- KafkaWriterState::getSubtaskId, Function.identity()));
- checkState(
- taskOffsetMapping.containsKey(subtaskId),
- "Internal error: It is expected that state from previous executions is distributed to the same subtask id.");
- final KafkaWriterState lastState = taskOffsetMapping.get(subtaskId);
- taskOffsetMapping.remove(subtaskId);
- abortTransactions(
- getTransactionsToAbort(lastState, new ArrayList<>(taskOffsetMapping.values())));
- if (!lastState.getTransactionalIdPrefix().equals(transactionalIdPrefix)) {
- LOG.warn(
- "Transactional id prefix from previous execution {} has changed to {}.",
- lastState.getTransactionalIdPrefix(),
- transactionalIdPrefix);
- return new KafkaWriterState(transactionalIdPrefix, subtaskId, 0);
+ private void abortLingeringTransactions(
+ List<KafkaWriterState> recoveredStates, long startCheckpointId) {
+ List<String> prefixesToAbort = Lists.newArrayList(transactionalIdPrefix);
+
+ if (!recoveredStates.isEmpty()) {
+ KafkaWriterState lastState = recoveredStates.get(0);
+ if (!lastState.getTransactionalIdPrefix().equals(transactionalIdPrefix)) {
+ prefixesToAbort.add(lastState.getTransactionalIdPrefix());
+ LOG.warn(
+ "Transactional id prefix from previous execution {} has changed to {}.",
+ lastState.getTransactionalIdPrefix(),
+ transactionalIdPrefix);
+ }
}
- return new KafkaWriterState(
- transactionalIdPrefix, subtaskId, lastState.getTransactionalIdOffset());
- }
- private void abortTransactions(List<String> transactionsToAbort) {
- transactionsToAbort.forEach(
- transaction -> {
- // don't mess with the original configuration or any other
- // properties of the
- // original object
- // -> create an internal kafka producer on our own and do not rely
- // on
- // initTransactionalProducer().
- LOG.info("Aborting Kafka transaction {}.", transaction);
- FlinkKafkaInternalProducer<byte[], byte[]> kafkaProducer = null;
- try {
- kafkaProducer =
- new FlinkKafkaInternalProducer<>(kafkaProducerConfig, transaction);
- // it suffices to call initTransactions - this will abort any
- // lingering transactions
- kafkaProducer.initTransactions();
- } finally {
- if (kafkaProducer != null) {
- kafkaProducer.close(Duration.ofSeconds(0));
- }
- }
- });
+ try (TransactionAborter transactionAborter =
+ new TransactionAborter(
+ kafkaSinkContext.getParallelInstanceId(),
+ kafkaSinkContext.getNumberOfParallelInstances(),
+ this::getOrCreateTransactionalProducer,
+ FlinkKafkaInternalProducer::close)) {
+ transactionAborter.abortLingeringTransactions(prefixesToAbort, startCheckpointId);
+ }
}
private void acknowledgeMessage() {
@@ -257,14 +225,13 @@ class KafkaWriter<IN> implements SinkWriter<IN, KafkaCommittable, KafkaWriterSta
}
}
- private FlinkKafkaInternalProducer<byte[], byte[]> createProducer() {
+ private FlinkKafkaInternalProducer<byte[], byte[]> createProducer(long checkpointId) {
switch (deliveryGuarantee) {
case EXACTLY_ONCE:
final FlinkKafkaInternalProducer<byte[], byte[]> transactionalProducer =
- createTransactionalProducer();
+ getTransactionalProducer(checkpointId);
initMetrics(transactionalProducer);
transactionalProducer.beginTransaction();
- closer.register(transactionalProducer);
return transactionalProducer;
case AT_LEAST_ONCE:
case NONE:
@@ -274,8 +241,8 @@ class KafkaWriter<IN> implements SinkWriter<IN, KafkaCommittable, KafkaWriterSta
}
final FlinkKafkaInternalProducer<byte[], byte[]> producer =
new FlinkKafkaInternalProducer<>(kafkaProducerConfig, null);
- initMetrics(producer);
closer.register(producer);
+ initMetrics(producer);
return producer;
default:
throw new UnsupportedOperationException(
@@ -330,23 +297,38 @@ class KafkaWriter<IN> implements SinkWriter<IN, KafkaCommittable, KafkaWriterSta
* For each checkpoint we create new {@link FlinkKafkaInternalProducer} so that new transactions
* will not clash with transactions created during previous checkpoints ({@code
* producer.initTransactions()} assures that we obtain new producerId and epoch counters).
+ *
+ * <p>Ensures that all transaction ids in between lastCheckpointId and checkpointId are
+ * initialized.
*/
- private FlinkKafkaInternalProducer<byte[], byte[]> createTransactionalProducer() {
- final long transactionalIdOffset = kafkaWriterState.getTransactionalIdOffset() + 1;
- String transactionalId =
- TransactionalIdFactory.buildTransactionalId(
- transactionalIdPrefix,
- kafkaSinkContext.getParallelInstanceId(),
- transactionalIdOffset);
- final FlinkKafkaInternalProducer<byte[], byte[]> producer =
+ private FlinkKafkaInternalProducer<byte[], byte[]> getTransactionalProducer(long checkpointId) {
+ checkState(
+ checkpointId > lastCheckpointId,
+ "Expected %s > %s",
+ checkpointId,
+ lastCheckpointId);
+ FlinkKafkaInternalProducer<byte[], byte[]> producer = null;
+ // in case checkpoints have been aborted, Flink would create non-consecutive transaction ids
+ // this loop ensures that all gaps are filled with initialized (empty) transactions
+ for (long id = lastCheckpointId + 1; id <= checkpointId; id++) {
+ String transactionalId =
+ TransactionalIdFactory.buildTransactionalId(
+ transactionalIdPrefix, kafkaSinkContext.getParallelInstanceId(), id);
+ producer = getOrCreateTransactionalProducer(transactionalId);
+ }
+ this.lastCheckpointId = checkpointId;
+ assert producer != null;
+ LOG.info("Created new transactional producer {}", producer.getTransactionalId());
+ return producer;
+ }
+
+ private FlinkKafkaInternalProducer<byte[], byte[]> getOrCreateTransactionalProducer(
+ String transactionalId) {
+ FlinkKafkaInternalProducer<byte[], byte[]> producer =
new FlinkKafkaInternalProducer<>(kafkaProducerConfig, transactionalId);
+ closer.register(producer);
producer.initTransactions();
- kafkaWriterState =
- new KafkaWriterState(
- transactionalIdPrefix,
- kafkaSinkContext.getParallelInstanceId(),
- transactionalIdOffset);
- LOG.info("Created new transactional producer {}", transactionalId);
+ initMetrics(producer);
return producer;
}
@@ -378,26 +360,6 @@ class KafkaWriter<IN> implements SinkWriter<IN, KafkaCommittable, KafkaWriterSta
};
}
- private List<String> getTransactionsToAbort(
- KafkaWriterState main, List<KafkaWriterState> others) {
- try (final KafkaTransactionLog log =
- new KafkaTransactionLog(
- kafkaProducerConfig,
- main,
- others,
- kafkaSinkContext.getNumberOfParallelInstances())) {
- return log.getTransactionsToAbort();
- } catch (KafkaException e) {
- LOG.warn(
- "Cannot abort transactions before startup e.g. the job has no access to the "
- + "__transaction_state topic. Lingering transactions may hold new "
- + "data back from downstream consumers. Please abort these "
- + "transactions manually.",
- e);
- return Collections.emptyList();
- }
- }
-
private static long computeSendTime(Producer<?, ?> producer) {
final Metric sendTime =
MetricUtil.getKafkaMetric(
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriterState.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriterState.java
index b5187b6..b4482c6 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriterState.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriterState.java
@@ -22,25 +22,12 @@ import java.util.Objects;
import static org.apache.flink.util.Preconditions.checkNotNull;
class KafkaWriterState {
- private final int subtaskId;
- private final long transactionalIdOffset;
private final String transactionalIdPrefix;
- KafkaWriterState(
- String transactionalIdPrefix, int subtaskId, long nextFreeTransactionalIdOffset) {
- this.subtaskId = subtaskId;
- this.transactionalIdOffset = nextFreeTransactionalIdOffset;
+ KafkaWriterState(String transactionalIdPrefix) {
this.transactionalIdPrefix = checkNotNull(transactionalIdPrefix, "transactionalIdPrefix");
}
- public int getSubtaskId() {
- return subtaskId;
- }
-
- public long getTransactionalIdOffset() {
- return transactionalIdOffset;
- }
-
public String getTransactionalIdPrefix() {
return transactionalIdPrefix;
}
@@ -54,21 +41,17 @@ class KafkaWriterState {
return false;
}
KafkaWriterState that = (KafkaWriterState) o;
- return subtaskId == that.subtaskId && transactionalIdOffset == that.transactionalIdOffset;
+ return transactionalIdPrefix.equals(that.transactionalIdPrefix);
}
@Override
public int hashCode() {
- return Objects.hash(subtaskId, transactionalIdOffset);
+ return Objects.hash(transactionalIdPrefix);
}
@Override
public String toString() {
return "KafkaWriterState{"
- + "subtaskId="
- + subtaskId
- + ", transactionalIdOffset="
- + transactionalIdOffset
+ ", transactionalIdPrefix='"
+ transactionalIdPrefix
+ '\''
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriterStateSerializer.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriterStateSerializer.java
index 609b899..5c91967 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriterStateSerializer.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriterStateSerializer.java
@@ -37,8 +37,6 @@ class KafkaWriterStateSerializer implements SimpleVersionedSerializer<KafkaWrite
public byte[] serialize(KafkaWriterState state) throws IOException {
try (final ByteArrayOutputStream baos = new ByteArrayOutputStream();
final DataOutputStream out = new DataOutputStream(baos)) {
- out.writeInt(state.getSubtaskId());
- out.writeLong(state.getTransactionalIdOffset());
out.writeUTF(state.getTransactionalIdPrefix());
out.flush();
return baos.toByteArray();
@@ -49,10 +47,8 @@ class KafkaWriterStateSerializer implements SimpleVersionedSerializer<KafkaWrite
public KafkaWriterState deserialize(int version, byte[] serialized) throws IOException {
try (final ByteArrayInputStream bais = new ByteArrayInputStream(serialized);
final DataInputStream in = new DataInputStream(bais)) {
- final int lastParallelism = in.readInt();
- final long transactionalOffset = in.readLong();
final String transactionalIdPrefx = in.readUTF();
- return new KafkaWriterState(transactionalIdPrefx, lastParallelism, transactionalOffset);
+ return new KafkaWriterState(transactionalIdPrefx);
}
}
}
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/TransactionAborter.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/TransactionAborter.java
new file mode 100644
index 0000000..cae6ca6
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/TransactionAborter.java
@@ -0,0 +1,131 @@
+/*
+ * 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.flink.connector.kafka.sink;
+
+import javax.annotation.Nullable;
+
+import java.io.Closeable;
+import java.util.List;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Aborts lingering transactions on restart.
+ *
+ * <p>Transactions are lingering if they are not tracked anywhere. For example, if a job is started
+ * transactions are opened. A restart without checkpoint would not allow Flink to abort old
+ * transactions. Since Kafka's transactions are sequential, newly produced data does not become
+ * visible for read_committed consumers. However, Kafka has no API for querying open transactions,
+ * so they become lingering.
+ *
+ * <p>Flink solves this by assuming consecutive transaction ids. On restart of checkpoint C on
+ * subtask S, it will sequentially cancel transaction C+1, C+2, ... of S until it finds the first
+ * unused transaction.
+ *
+ * <p>Additionally, to cover for weird downscaling cases without checkpoints, it also checks for
+ * transactions of subtask S+P where P is the current parallelism until it finds a subtask without
+ * transactions.
+ */
+class TransactionAborter implements Closeable {
+ private final int subtaskId;
+ private final int parallelism;
+ private final Function<String, FlinkKafkaInternalProducer<byte[], byte[]>> producerFactory;
+ private final Consumer<FlinkKafkaInternalProducer<byte[], byte[]>> closeAction;
+ @Nullable FlinkKafkaInternalProducer<byte[], byte[]> producer = null;
+
+ public TransactionAborter(
+ int subtaskId,
+ int parallelism,
+ Function<String, FlinkKafkaInternalProducer<byte[], byte[]>> producerFactory,
+ Consumer<FlinkKafkaInternalProducer<byte[], byte[]>> closeAction) {
+ this.subtaskId = subtaskId;
+ this.parallelism = parallelism;
+ this.producerFactory = checkNotNull(producerFactory);
+ this.closeAction = closeAction;
+ }
+
+ void abortLingeringTransactions(List<String> prefixesToAbort, long startCheckpointId) {
+ for (String prefix : prefixesToAbort) {
+ abortTransactionsWithPrefix(prefix, startCheckpointId);
+ }
+ }
+
+ /**
+ * Aborts all transactions that have been created by this subtask in a previous run.
+ *
+ * <p>It also aborts transactions from subtasks that may have been removed because of
+ * downscaling.
+ *
+ * <p>When Flink downscales X subtasks to Y subtasks, then subtask i is responsible for cleaning
+ * all subtasks j in [0; X), where j % Y = i. For example, if we downscale to 2, then subtask 0
+ * is responsible for all even and subtask 1 for all odd subtasks.
+ */
+ private void abortTransactionsWithPrefix(String prefix, long startCheckpointId) {
+ for (int subtaskId = this.subtaskId; ; subtaskId += parallelism) {
+ if (abortTransactionOfSubtask(prefix, startCheckpointId, subtaskId) == 0) {
+ // If Flink didn't abort any transaction for current subtask, then we assume that no
+ // such subtask existed and no subtask with a higher number as well.
+ break;
+ }
+ }
+ }
+
+ /**
+ * Aborts all transactions that have been created by a subtask in a previous run after the given
+ * checkpoint id.
+ *
+ * <p>We assume that transaction ids are consecutively used and thus Flink can stop aborting as
+ * soon as Flink notices that a particular transaction id was unused.
+ */
+ private int abortTransactionOfSubtask(String prefix, long startCheckpointId, int subtaskId) {
+ int numTransactionAborted = 0;
+ for (long checkpointId = startCheckpointId; ; checkpointId++, numTransactionAborted++) {
+ // initTransactions fences all old transactions with the same id by bumping the epoch
+ String transactionalId =
+ TransactionalIdFactory.buildTransactionalId(prefix, subtaskId, checkpointId);
+ if (producer == null) {
+ producer = producerFactory.apply(transactionalId);
+ } else {
+ producer.initTransactionId(transactionalId);
+ }
+ producer.flush();
+ // An epoch of 0 indicates that the id was unused before
+ if (producer.getEpoch() == 0) {
+ // Note that the check works beyond transaction log timeouts and just depends on the
+ // retention of the transaction topic (typically 7d). Any transaction that is not in
+ // the that topic anymore is also not lingering (i.e., it will not block downstream
+ // from reading)
+ // This method will only cease to work if transaction log timeout = topic retention
+ // and a user didn't restart the application for that period of time. Then the first
+ // transactions would vanish from the topic while later transactions are still
+ // lingering until they are cleaned up by Kafka. Then the user has to wait until the
+ // other transactions are timed out (which shouldn't take too long).
+ break;
+ }
+ }
+ return numTransactionAborted;
+ }
+
+ public void close() {
+ if (producer != null) {
+ closeAction.accept(producer);
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/TransactionalIdFactory.java b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/TransactionalIdFactory.java
index 96744b6..eda4c01 100644
--- a/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/TransactionalIdFactory.java
+++ b/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/TransactionalIdFactory.java
@@ -17,17 +17,7 @@
package org.apache.flink.connector.kafka.sink;
-import org.apache.flink.shaded.guava30.com.google.common.base.Splitter;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Optional;
-
class TransactionalIdFactory {
- private static final Logger LOG = LoggerFactory.getLogger(TransactionalIdFactory.class);
private static final String TRANSACTIONAL_ID_DELIMITER = "-";
/**
@@ -48,36 +38,4 @@ class TransactionalIdFactory {
+ TRANSACTIONAL_ID_DELIMITER
+ checkpointOffset;
}
-
- /**
- * Tries to parse the {@link KafkaWriterState} which was used to create the transactionalId.
- *
- * @param transactionalId read from the transaction topic
- * @return returns {@link KafkaWriterState} if the transaction was create by the Kafka Sink.
- */
- public static Optional<KafkaWriterState> parseKafkaWriterState(String transactionalId) {
- final List<String> splits = new ArrayList<>();
- Splitter.on(TRANSACTIONAL_ID_DELIMITER).split(transactionalId).forEach(splits::add);
- final int splitSize = splits.size();
- if (splitSize < 3) {
- LOG.debug("Transaction {} was not created by the Flink Kafka sink", transactionalId);
- return Optional.empty();
- }
- try {
- final long checkpointOffset = Long.parseLong(splits.get(splitSize - 1));
- final int subtaskId = Integer.parseInt(splits.get(splitSize - 2));
- return Optional.of(
- new KafkaWriterState(
- String.join(
- TRANSACTIONAL_ID_DELIMITER, splits.subList(0, splitSize - 2)),
- subtaskId,
- checkpointOffset));
- } catch (NumberFormatException e) {
- LOG.debug(
- "Transaction {} was not created by the Flink Kafka sink: {}",
- transactionalId,
- e);
- return Optional.empty();
- }
- }
}
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducerITCase.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducerITCase.java
new file mode 100644
index 0000000..585303c
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducerITCase.java
@@ -0,0 +1,122 @@
+/*
+ * 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.flink.connector.kafka.sink;
+
+import org.apache.flink.util.TestLogger;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Properties;
+import java.util.stream.Collectors;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasSize;
+
+@Testcontainers
+class FlinkKafkaInternalProducerITCase extends TestLogger {
+
+ private static final String TEST_TOPIC = "test-topic";
+ private static final Logger LOG =
+ LoggerFactory.getLogger(FlinkKafkaInternalProducerITCase.class);
+
+ @Container
+ private static final KafkaContainer KAFKA_CONTAINER =
+ new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:5.5.2"))
+ .withEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "1")
+ .withEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "1")
+ .withLogConsumer(new Slf4jLogConsumer(LOG))
+ .withEmbeddedZookeeper();
+
+ private static final String TRANSACTION_PREFIX = "test-transaction-";
+
+ Properties getProperties() {
+ Properties properties = new Properties();
+ properties.put(
+ CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
+ KAFKA_CONTAINER.getBootstrapServers());
+ properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
+ properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+ properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+ properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+ properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+ return properties;
+ }
+
+ @Test
+ void testInitTransactionId() {
+ try (FlinkKafkaInternalProducer<String, String> reuse =
+ new FlinkKafkaInternalProducer<>(getProperties(), "dummy")) {
+ int numTransactions = 20;
+ for (int i = 1; i <= numTransactions; i++) {
+ reuse.initTransactionId(TRANSACTION_PREFIX + i);
+ reuse.beginTransaction();
+ reuse.send(new ProducerRecord<>(TEST_TOPIC, "test-value-" + i));
+ if (i % 2 == 0) {
+ reuse.commitTransaction();
+ } else {
+ reuse.flush();
+ reuse.abortTransaction();
+ }
+ assertNumTransactions(i);
+ assertThat(readRecords(TEST_TOPIC).count(), equalTo(i / 2));
+ }
+ }
+ }
+
+ private void assertNumTransactions(int numTransactions) {
+ List<KafkaTransactionLog.TransactionRecord> transactions =
+ new KafkaTransactionLog(getProperties())
+ .getTransactions(id -> id.startsWith(TRANSACTION_PREFIX));
+ assertThat(
+ transactions.stream()
+ .map(KafkaTransactionLog.TransactionRecord::getTransactionId)
+ .collect(Collectors.toSet()),
+ hasSize(numTransactions));
+ }
+
+ private ConsumerRecords<String, String> readRecords(String topic) {
+ Properties properties = getProperties();
+ properties.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
+ KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
+ consumer.assign(
+ consumer.partitionsFor(topic).stream()
+ .map(partitionInfo -> new TopicPartition(topic, partitionInfo.partition()))
+ .collect(Collectors.toSet()));
+ consumer.seekToBeginning(consumer.assignment());
+ return consumer.poll(Duration.ofMillis(1000));
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaTransactionLog.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaTransactionLog.java
new file mode 100644
index 0000000..0526120
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaTransactionLog.java
@@ -0,0 +1,247 @@
+/*
+ * 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.flink.connector.kafka.sink;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.kafka.common.internals.Topic.TRANSACTION_STATE_TOPIC_NAME;
+
+/**
+ * This class is responsible to provide the format of the used transationalIds and in case of an
+ * application restart query the open transactions and decide which must be aborted.
+ */
+class KafkaTransactionLog {
+
+ private static final Duration CONSUMER_POLL_DURATION = Duration.ofSeconds(1);
+ private static final int SUPPORTED_KAFKA_SCHEMA_VERSION = 0;
+ private final Properties consumerConfig;
+
+ /**
+ * Constructor creating a KafkaTransactionLog.
+ *
+ * @param kafkaConfig used to configure the {@link KafkaConsumer} to query the topic containing
+ * the transaction information
+ */
+ KafkaTransactionLog(Properties kafkaConfig) {
+ this.consumerConfig = new Properties();
+ consumerConfig.putAll(checkNotNull(kafkaConfig, "kafkaConfig"));
+ consumerConfig.put("key.deserializer", ByteArrayDeserializer.class.getName());
+ consumerConfig.put("value.deserializer", ByteArrayDeserializer.class.getName());
+ consumerConfig.put("enable.auto.commit", false);
+ }
+
+ public List<TransactionRecord> getTransactions() {
+ return getTransactions(id -> true);
+ }
+
+ /** Gets all {@link TransactionRecord} matching the given id filter. */
+ public List<TransactionRecord> getTransactions(Predicate<String> transactionIdFilter)
+ throws KafkaException {
+ try (KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(consumerConfig)) {
+ Set<TopicPartition> assignments = getAllPartitions(consumer);
+ Map<TopicPartition, Long> endOffsets = consumer.endOffsets(assignments);
+ final Map<Integer, Long> endOffsetByPartition =
+ endOffsets.entrySet().stream()
+ .collect(
+ Collectors.toMap(
+ e -> e.getKey().partition(), Map.Entry::getValue));
+ consumer.beginningOffsets(assignments)
+ .forEach(
+ (tp, offset) -> {
+ if (endOffsets.get(tp) <= offset) {
+ assignments.remove(tp);
+ }
+ });
+ consumer.assign(assignments);
+ consumer.seekToBeginning(assignments);
+
+ final List<TransactionRecord> transactionRecords = new ArrayList<>();
+ while (!assignments.isEmpty()) {
+ ConsumerRecords<byte[], byte[]> records = consumer.poll(CONSUMER_POLL_DURATION);
+ boolean finishedPartition = false;
+ for (ConsumerRecord<byte[], byte[]> r : records) {
+ long remainingRecords = endOffsetByPartition.get(r.partition()) - r.offset();
+ if (remainingRecords >= 1) {
+ parseTransaction(r, transactionIdFilter).ifPresent(transactionRecords::add);
+ }
+ if (remainingRecords <= 1) {
+ assignments.remove(new TopicPartition(r.topic(), r.partition()));
+ finishedPartition = true;
+ }
+ }
+ if (finishedPartition) {
+ consumer.assign(assignments);
+ }
+ }
+ return transactionRecords;
+ }
+ }
+
+ private Set<TopicPartition> getAllPartitions(KafkaConsumer<byte[], byte[]> consumer) {
+ final List<PartitionInfo> partitionInfos =
+ consumer.partitionsFor(TRANSACTION_STATE_TOPIC_NAME);
+ return partitionInfos.stream()
+ .map(info -> new TopicPartition(info.topic(), info.partition()))
+ .collect(Collectors.toCollection(HashSet::new));
+ }
+
+ private Optional<TransactionRecord> parseTransaction(
+ ConsumerRecord<byte[], byte[]> consumerRecord, Predicate<String> transactionIdFilter) {
+ final ByteBuffer keyBuffer = ByteBuffer.wrap(consumerRecord.key());
+ checkKafkaSchemaVersionMatches(keyBuffer);
+ // Ignore 2 bytes because Kafka's internal representation
+ keyBuffer.getShort();
+ final String transactionalId = StandardCharsets.UTF_8.decode(keyBuffer).toString();
+
+ if (!transactionIdFilter.test(transactionalId)) {
+ return Optional.empty();
+ }
+
+ final ByteBuffer valueBuffer = ByteBuffer.wrap(consumerRecord.value());
+ checkKafkaSchemaVersionMatches(valueBuffer);
+ final TransactionState state = TransactionState.fromByte(readTransactionState(valueBuffer));
+
+ return Optional.of(new TransactionRecord(transactionalId, state));
+ }
+
+ private static byte readTransactionState(ByteBuffer buffer) {
+ // producerId
+ buffer.getLong();
+ // epoch
+ buffer.getShort();
+ // transactionTimeout
+ buffer.getInt();
+ // statusKey
+ return buffer.get();
+ }
+
+ public static class TransactionRecord {
+ private final String transactionId;
+ private final TransactionState state;
+
+ public TransactionRecord(String transactionId, TransactionState state) {
+ this.transactionId = checkNotNull(transactionId);
+ this.state = checkNotNull(state);
+ }
+
+ public String getTransactionId() {
+ return transactionId;
+ }
+
+ public TransactionState getState() {
+ return state;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ TransactionRecord that = (TransactionRecord) o;
+ return transactionId.equals(that.transactionId) && state == that.state;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(transactionId, state);
+ }
+
+ @Override
+ public String toString() {
+ return "TransactionRecord{"
+ + "transactionId='"
+ + transactionId
+ + '\''
+ + ", state="
+ + state
+ + '}';
+ }
+ }
+
+ public enum TransactionState {
+ Empty(Byte.parseByte("0"), false),
+ Ongoing(Byte.parseByte("1"), false),
+ PrepareCommit(Byte.parseByte("2"), false),
+ PrepareAbort(Byte.parseByte("3"), false),
+ CompleteCommit(Byte.parseByte("4"), true),
+ CompleteAbort(Byte.parseByte("5"), true),
+ Dead(Byte.parseByte("6"), true),
+ PrepareEpochFence(Byte.parseByte("7"), false);
+
+ private static final Map<Byte, TransactionState> BYTE_TO_STATE =
+ Arrays.stream(TransactionState.values())
+ .collect(Collectors.toMap(e -> e.state, e -> e));
+
+ private final byte state;
+
+ private boolean terminal;
+
+ TransactionState(byte state, boolean terminal) {
+ this.state = state;
+ this.terminal = terminal;
+ }
+
+ public boolean isTerminal() {
+ return terminal;
+ }
+
+ static TransactionState fromByte(byte state) {
+ final TransactionState transactionState = BYTE_TO_STATE.get(state);
+ if (transactionState == null) {
+ throw new IllegalArgumentException(
+ String.format("The given state %s is not supported.", state));
+ }
+ return transactionState;
+ }
+ }
+
+ private static void checkKafkaSchemaVersionMatches(ByteBuffer buffer) {
+ final short version = buffer.getShort();
+ if (version != SUPPORTED_KAFKA_SCHEMA_VERSION) {
+ throw new IllegalStateException(
+ String.format(
+ "Kafka has changed the schema version from %s to %s",
+ SUPPORTED_KAFKA_SCHEMA_VERSION, version));
+ }
+ }
+}
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaTransactionLogITCase.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaTransactionLogITCase.java
index a404783..4c9c802 100644
--- a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaTransactionLogITCase.java
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaTransactionLogITCase.java
@@ -17,11 +17,9 @@
package org.apache.flink.connector.kafka.sink;
+import org.apache.flink.connector.kafka.sink.KafkaTransactionLog.TransactionRecord;
import org.apache.flink.util.TestLogger;
-import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList;
-import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableMap;
-
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
@@ -34,7 +32,6 @@ import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.KafkaContainer;
-import org.testcontainers.containers.Network;
import org.testcontainers.containers.output.Slf4jLogConsumer;
import org.testcontainers.utility.DockerImageName;
@@ -44,6 +41,12 @@ import java.util.List;
import java.util.Properties;
import java.util.function.Consumer;
+import static org.apache.flink.connector.kafka.sink.KafkaTransactionLog.TransactionState.CompleteAbort;
+import static org.apache.flink.connector.kafka.sink.KafkaTransactionLog.TransactionState.CompleteCommit;
+import static org.apache.flink.connector.kafka.sink.KafkaTransactionLog.TransactionState.Empty;
+import static org.apache.flink.connector.kafka.sink.KafkaTransactionLog.TransactionState.Ongoing;
+import static org.apache.flink.connector.kafka.sink.KafkaTransactionLog.TransactionState.PrepareAbort;
+import static org.apache.flink.connector.kafka.sink.KafkaTransactionLog.TransactionState.PrepareCommit;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsInAnyOrder;
@@ -52,8 +55,6 @@ public class KafkaTransactionLogITCase extends TestLogger {
private static final Logger LOG = LoggerFactory.getLogger(KafkaSinkITCase.class);
private static final Slf4jLogConsumer LOG_CONSUMER = new Slf4jLogConsumer(LOG);
- private static final String INTER_CONTAINER_KAFKA_ALIAS = "kafka";
- private static final Network NETWORK = Network.newNetwork();
private static final String TOPIC_NAME = "kafkaTransactionLogTest";
private static final String TRANSACTIONAL_ID_PREFIX = "kafka-log";
@@ -61,19 +62,13 @@ public class KafkaTransactionLogITCase extends TestLogger {
public static final KafkaContainer KAFKA_CONTAINER =
new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:6.2.0"))
.withEmbeddedZookeeper()
+ .withEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "1")
+ .withEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "1")
+ .withEnv("KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE", "false")
.withEnv(
- ImmutableMap.of(
- "KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR",
- "1",
- "KAFKA_TRANSACTION_MAX_TIMEOUT_MS",
- String.valueOf(Duration.ofHours(2).toMillis()),
- "KAFKA_TRANSACTION_STATE_LOG_MIN_ISR",
- "1",
- "KAFKA_MIN_INSYNC_REPLICAS",
- "1"))
- .withNetwork(NETWORK)
- .withLogConsumer(LOG_CONSUMER)
- .withNetworkAliases(INTER_CONTAINER_KAFKA_ALIAS);
+ "KAFKA_TRANSACTION_MAX_TIMEOUT_MS",
+ String.valueOf(Duration.ofHours(2).toMillis()))
+ .withLogConsumer(LOG_CONSUMER);
private final List<Producer<byte[], Integer>> openProducers = new ArrayList<>();
@@ -83,64 +78,35 @@ public class KafkaTransactionLogITCase extends TestLogger {
}
@Test
- public void testGetTransactionsToAbort() {
- committedTransaction(0, 1);
- abortedTransaction(0, 2);
- lingeringTransaction(0, 3);
- lingeringTransaction(0, 4);
-
- committedTransaction(2, 1);
- lingeringTransaction(2, 2);
-
- committedTransaction(3, 1);
- lingeringTransaction(3, 2);
-
- committedTransaction(5, 1);
- lingeringTransaction(5, 2);
-
- committedTransaction(6, 1);
-
- lingeringTransaction(7, 1);
- lingeringTransaction(8, 1);
-
- try (final KafkaTransactionLog transactionLog =
- new KafkaTransactionLog(
- getKafkaClientConfiguration(),
- createWriterState(0, 1),
- ImmutableList.of(createWriterState(5, 1), createWriterState(2, 0)),
- 2)) {
- final List<String> transactionsToAbort = transactionLog.getTransactionsToAbort();
- assertThat(
- transactionsToAbort,
- containsInAnyOrder(
- buildTransactionalId(0, 3),
- buildTransactionalId(0, 4),
- buildTransactionalId(2, 2),
- buildTransactionalId(5, 2),
- buildTransactionalId(8, 1)));
- }
-
- try (final KafkaTransactionLog transactionLog =
- new KafkaTransactionLog(
- getKafkaClientConfiguration(),
- createWriterState(1, 1),
- ImmutableList.of(createWriterState(6, 1), createWriterState(3, 1)),
- 2)) {
- final List<String> transactionsToAbort = transactionLog.getTransactionsToAbort();
- assertThat(
- transactionsToAbort,
- containsInAnyOrder(buildTransactionalId(3, 2), buildTransactionalId(7, 1)));
- }
- }
-
- private static KafkaWriterState createWriterState(int subtaskId, long offset) {
- return new KafkaWriterState(TRANSACTIONAL_ID_PREFIX, subtaskId, offset);
+ public void testGetTransactions() {
+ committedTransaction(1);
+ abortedTransaction(2);
+ lingeringTransaction(3);
+ lingeringTransaction(4);
+
+ final KafkaTransactionLog transactionLog =
+ new KafkaTransactionLog(getKafkaClientConfiguration());
+ final List<TransactionRecord> transactions = transactionLog.getTransactions();
+ assertThat(
+ transactions,
+ containsInAnyOrder(
+ new TransactionRecord(buildTransactionalId(1), Empty),
+ new TransactionRecord(buildTransactionalId(1), Ongoing),
+ new TransactionRecord(buildTransactionalId(1), PrepareCommit),
+ new TransactionRecord(buildTransactionalId(1), CompleteCommit),
+ new TransactionRecord(buildTransactionalId(2), Empty),
+ new TransactionRecord(buildTransactionalId(2), Ongoing),
+ new TransactionRecord(buildTransactionalId(2), PrepareAbort),
+ new TransactionRecord(buildTransactionalId(2), CompleteAbort),
+ new TransactionRecord(buildTransactionalId(3), Empty),
+ new TransactionRecord(buildTransactionalId(3), Ongoing),
+ new TransactionRecord(buildTransactionalId(4), Empty),
+ new TransactionRecord(buildTransactionalId(4), Ongoing)));
}
- private void committedTransaction(int subtaskId, long offset) {
+ private void committedTransaction(long id) {
submitTransaction(
- subtaskId,
- offset,
+ id,
producer -> {
producer.initTransactions();
producer.beginTransaction();
@@ -151,10 +117,9 @@ public class KafkaTransactionLogITCase extends TestLogger {
});
}
- private void lingeringTransaction(int subtaskId, long offset) {
+ private void lingeringTransaction(long id) {
submitTransaction(
- subtaskId,
- offset,
+ id,
producer -> {
producer.initTransactions();
producer.beginTransaction();
@@ -163,10 +128,9 @@ public class KafkaTransactionLogITCase extends TestLogger {
});
}
- private void abortedTransaction(int subtaskId, long offset) {
+ private void abortedTransaction(long id) {
submitTransaction(
- subtaskId,
- offset,
+ id,
producer -> {
producer.initTransactions();
producer.beginTransaction();
@@ -177,17 +141,15 @@ public class KafkaTransactionLogITCase extends TestLogger {
});
}
- private void submitTransaction(
- int subtaskId, long offset, Consumer<Producer<byte[], Integer>> producerAction) {
- final Producer<byte[], Integer> producer =
- createProducer(buildTransactionalId(subtaskId, offset));
+ private void submitTransaction(long id, Consumer<Producer<byte[], Integer>> producerAction) {
+ Producer<byte[], Integer> producer = createProducer(buildTransactionalId(id));
openProducers.add(producer);
producerAction.accept(producer);
+ // don't close here for lingering transactions
}
- private static String buildTransactionalId(int subtaskId, long offset) {
- return TransactionalIdFactory.buildTransactionalId(
- TRANSACTIONAL_ID_PREFIX, subtaskId, offset);
+ private static String buildTransactionalId(long id) {
+ return TRANSACTIONAL_ID_PREFIX + id;
}
private static Producer<byte[], Integer> createProducer(String transactionalId) {
@@ -205,7 +167,7 @@ public class KafkaTransactionLogITCase extends TestLogger {
standardProps.put("bootstrap.servers", KAFKA_CONTAINER.getBootstrapServers());
standardProps.put("group.id", "flink-tests");
standardProps.put("enable.auto.commit", false);
- standardProps.put("auto.offset.reset", "earliest");
+ standardProps.put("auto.id.reset", "earliest");
standardProps.put("max.partition.fetch.bytes", 256);
return standardProps;
}
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterStateSerializerTest.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterStateSerializerTest.java
index a96f033..afdd430 100644
--- a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterStateSerializerTest.java
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaWriterStateSerializerTest.java
@@ -35,7 +35,7 @@ public class KafkaWriterStateSerializerTest extends TestLogger {
@Test
public void testStateSerDe() throws IOException {
- final KafkaWriterState state = new KafkaWriterState("idPrefix", 1, 2);
+ final KafkaWriterState state = new KafkaWriterState("idPrefix");
final byte[] serialized = SERIALIZER.serialize(state);
assertEquals(state, SERIALIZER.deserialize(1, serialized));
}
diff --git a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/TransactionIdFactoryTest.java b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/TransactionIdFactoryTest.java
index 5b36a6c..e54d0a3 100644
--- a/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/TransactionIdFactoryTest.java
+++ b/flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/TransactionIdFactoryTest.java
@@ -21,11 +21,7 @@ import org.apache.flink.util.TestLogger;
import org.junit.Test;
-import java.util.Optional;
-
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
/** Tests for {@link TransactionalIdFactory}. */
public class TransactionIdFactoryTest extends TestLogger {
@@ -35,40 +31,4 @@ public class TransactionIdFactoryTest extends TestLogger {
final String expected = "prefix-0-2";
assertEquals(expected, TransactionalIdFactory.buildTransactionalId("prefix", 0, 2L));
}
-
- @Test
- public void testParseStateFromIdWithNotEnoughDelimiters() {
- final String transactionalId = "prefix-0";
- assertFalse(TransactionalIdFactory.parseKafkaWriterState(transactionalId).isPresent());
- }
-
- @Test
- public void testParseStateFromId() {
- final String transactionalId = "prefix-0-2";
- final Optional<KafkaWriterState> stateOpt =
- TransactionalIdFactory.parseKafkaWriterState(transactionalId);
- assertTrue(stateOpt.isPresent());
- assertEquals(new KafkaWriterState("prefix", 0, 2L), stateOpt.get());
- }
-
- @Test
- public void testParseStateFromIdWithPrefixContainingDelimiter() {
- final String transactionalId = "prefix1-prefix2-0-2";
- final Optional<KafkaWriterState> stateOpt =
- TransactionalIdFactory.parseKafkaWriterState(transactionalId);
- assertTrue(stateOpt.isPresent());
- assertEquals(new KafkaWriterState("prefix1-prefix-2", 0, 2L), stateOpt.get());
- }
-
- @Test
- public void testParseStateFromIdWithInvalidSubtaskId() {
- final String transactionalId = "prefix-invalid-2";
- assertFalse(TransactionalIdFactory.parseKafkaWriterState(transactionalId).isPresent());
- }
-
- @Test
- public void testParseStateFromIdWithInvalidCheckpointOffset() {
- final String transactionalId = "prefix-0-invalid";
- assertFalse(TransactionalIdFactory.parseKafkaWriterState(transactionalId).isPresent());
- }
}