flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From aljos...@apache.org
Subject [2/2] flink git commit: [hotfix][kafka] Extract TransactionalIdsGenerator class from FlinkKafkaProducer011
Date Tue, 07 Nov 2017 18:01:10 GMT
[hotfix][kafka] Extract TransactionalIdsGenerator class from FlinkKafkaProducer011

This is pure refactor without any functional changes.


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/3cbf467e
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/3cbf467e
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/3cbf467e

Branch: refs/heads/release-1.4
Commit: 3cbf467ebdf639df4d7d4da78b7bc2929aa4b5d9
Parents: 564c993
Author: Piotr Nowojski <piotr.nowojski@gmail.com>
Authored: Mon Nov 6 14:03:16 2017 +0100
Committer: Aljoscha Krettek <aljoscha.krettek@gmail.com>
Committed: Tue Nov 7 18:58:12 2017 +0100

----------------------------------------------------------------------
 .../connectors/kafka/FlinkKafkaProducer011.java | 53 +++++--------
 .../internal/TransactionalIdsGenerator.java     | 81 ++++++++++++++++++++
 2 files changed, 99 insertions(+), 35 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/3cbf467e/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java
b/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java
index 873ef08..0310019 100644
--- a/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java
+++ b/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java
@@ -37,6 +37,7 @@ import org.apache.flink.runtime.state.FunctionSnapshotContext;
 import org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction;
 import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
 import org.apache.flink.streaming.connectors.kafka.internal.FlinkKafkaProducer;
+import org.apache.flink.streaming.connectors.kafka.internal.TransactionalIdsGenerator;
 import org.apache.flink.streaming.connectors.kafka.internal.metrics.KafkaMetricMuttableWrapper;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkFixedPartitioner;
 import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaDelegatePartitioner;
@@ -59,7 +60,6 @@ import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.errors.InvalidTxnStateException;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -81,8 +81,6 @@ import java.util.Set;
 import java.util.concurrent.BlockingDeque;
 import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.atomic.AtomicLong;
-import java.util.stream.LongStream;
-import java.util.stream.Stream;
 
 import static org.apache.flink.util.Preconditions.checkNotNull;
 import static org.apache.flink.util.Preconditions.checkState;
@@ -183,6 +181,11 @@ public class FlinkKafkaProducer011<IN>
 	private transient ListState<NextTransactionalIdHint> nextTransactionalIdHintState;
 
 	/**
+	 * Generator for Transactional IDs.
+	 */
+	private transient TransactionalIdsGenerator transactionalIdsGenerator;
+
+	/**
 	 * Hint for picking next transactional id.
 	 */
 	private transient NextTransactionalIdHint nextTransactionalIdHint;
@@ -785,6 +788,11 @@ public class FlinkKafkaProducer011<IN>
 
 		nextTransactionalIdHintState = context.getOperatorStateStore().getUnionListState(
 			NEXT_TRANSACTIONAL_ID_HINT_DESCRIPTOR);
+		transactionalIdsGenerator = new TransactionalIdsGenerator(
+			getRuntimeContext().getTaskName(),
+			getRuntimeContext().getIndexOfThisSubtask(),
+			kafkaProducersPoolSize,
+			SAFE_SCALE_DOWN_FACTOR);
 
 		if (semantic != Semantic.EXACTLY_ONCE) {
 			nextTransactionalIdHint = null;
@@ -803,15 +811,8 @@ public class FlinkKafkaProducer011<IN>
 				// (1) the first execution of this application
 				// (2) previous execution has failed before first checkpoint completed
 				//
-				// in case of (2) we have to abort all previous transactions, but we don't know was the
parallelism used
-				// then, so we must guess using current configured pool size, current parallelism and
-				// SAFE_SCALE_DOWN_FACTOR
-				long abortTransactionalIdStart = getRuntimeContext().getIndexOfThisSubtask();
-				long abortTransactionalIdEnd = abortTransactionalIdStart + 1;
-
-				abortTransactionalIdStart *= kafkaProducersPoolSize * SAFE_SCALE_DOWN_FACTOR;
-				abortTransactionalIdEnd *= kafkaProducersPoolSize * SAFE_SCALE_DOWN_FACTOR;
-				abortTransactions(LongStream.range(abortTransactionalIdStart, abortTransactionalIdEnd));
+				// in case of (2) we have to abort all previous transactions
+				abortTransactions(transactionalIdsGenerator.generateIdsToAbort());
 			} else {
 				nextTransactionalIdHint = transactionalIdHints.get(0);
 			}
@@ -834,16 +835,7 @@ public class FlinkKafkaProducer011<IN>
 	private Set<String> generateNewTransactionalIds() {
 		checkState(nextTransactionalIdHint != null, "nextTransactionalIdHint must be present for
EXACTLY_ONCE");
 
-		// range of available transactional ids is:
-		// [nextFreeTransactionalId, nextFreeTransactionalId + parallelism * kafkaProducersPoolSize)
-		// loop below picks in a deterministic way a subrange of those available transactional
ids based on index of
-		// this subtask
-		int subtaskId = getRuntimeContext().getIndexOfThisSubtask();
-		Set<String> transactionalIds = new HashSet<>();
-		for (int i = 0; i < kafkaProducersPoolSize; i++) {
-			long transactionalId = nextTransactionalIdHint.nextFreeTransactionalId + subtaskId * kafkaProducersPoolSize
+ i;
-			transactionalIds.add(generateTransactionalId(transactionalId));
-		}
+		Set<String> transactionalIds = transactionalIdsGenerator.generateIdsToUse(nextTransactionalIdHint.nextFreeTransactionalId);
 		LOG.info("Generated new transactionalIds {}", transactionalIds);
 		return transactionalIds;
 	}
@@ -862,7 +854,7 @@ public class FlinkKafkaProducer011<IN>
 		if (!getUserContext().isPresent()) {
 			return;
 		}
-		abortTransactions(getUserContext().get().transactionalIds.stream());
+		abortTransactions(getUserContext().get().transactionalIds);
 	}
 
 	private void resetAvailableTransactionalIdsPool(Collection<String> transactionalIds)
{
@@ -874,22 +866,13 @@ public class FlinkKafkaProducer011<IN>
 
 	// ----------------------------------- Utilities --------------------------
 
-	private void abortTransactions(LongStream transactionalIds) {
-		abortTransactions(transactionalIds.mapToObj(this::generateTransactionalId));
-	}
-
-	private void abortTransactions(Stream<String> transactionalIds) {
-		transactionalIds.forEach(transactionalId -> {
+	private void abortTransactions(Set<String> transactionalIds) {
+		for (String transactionalId : transactionalIds) {
 			try (FlinkKafkaProducer<byte[], byte[]> kafkaProducer =
 					initTransactionalProducer(transactionalId, false)) {
 				kafkaProducer.initTransactions();
 			}
-		});
-	}
-
-	private String generateTransactionalId(long transactionalId) {
-		String transactionalIdFormat = getRuntimeContext().getTaskName() + "-%d";
-		return String.format(transactionalIdFormat, transactionalId);
+		}
 	}
 
 	int getTransactionCoordinatorId() {

http://git-wip-us.apache.org/repos/asf/flink/blob/3cbf467e/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/TransactionalIdsGenerator.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/TransactionalIdsGenerator.java
b/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/TransactionalIdsGenerator.java
new file mode 100644
index 0000000..eabdad9
--- /dev/null
+++ b/flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/TransactionalIdsGenerator.java
@@ -0,0 +1,81 @@
+/*
+ * 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.streaming.connectors.kafka.internal;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.LongStream;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Class responsible for generating transactional ids to use when communicating with Kafka.
+ */
+public class TransactionalIdsGenerator {
+	private final String prefix;
+	private final int subtaskIndex;
+	private final int poolSize;
+	private final int safeScaleDownFactor;
+
+	public TransactionalIdsGenerator(
+			String prefix,
+			int subtaskIndex,
+			int poolSize,
+			int safeScaleDownFactor) {
+		this.prefix = checkNotNull(prefix);
+		this.subtaskIndex = subtaskIndex;
+		this.poolSize = poolSize;
+		this.safeScaleDownFactor = safeScaleDownFactor;
+	}
+
+	/**
+	 * Range of available transactional ids to use is:
+	 * [nextFreeTransactionalId, nextFreeTransactionalId + parallelism * kafkaProducersPoolSize)
+	 * loop below picks in a deterministic way a subrange of those available transactional ids
based on index of
+	 * this subtask.
+	 */
+	public Set<String> generateIdsToUse(long nextFreeTransactionalId) {
+		Set<String> transactionalIds = new HashSet<>();
+		for (int i = 0; i < poolSize; i++) {
+			long transactionalId = nextFreeTransactionalId + subtaskIndex * poolSize + i;
+			transactionalIds.add(generateTransactionalId(transactionalId));
+		}
+		return transactionalIds;
+	}
+
+	/**
+	 *  If we have to abort previous transactional id in case of restart after a failure BEFORE
first checkpoint
+	 *  completed, we don't know what was the parallelism used in previous attempt. In that
case we must guess the ids
+	 *  range to abort based on current configured pool size, current parallelism and safeScaleDownFactor.
+	 */
+	public Set<String> generateIdsToAbort() {
+		long abortTransactionalIdStart = subtaskIndex;
+		long abortTransactionalIdEnd = abortTransactionalIdStart + 1;
+
+		abortTransactionalIdStart *= poolSize * safeScaleDownFactor;
+		abortTransactionalIdEnd *= poolSize * safeScaleDownFactor;
+		return LongStream.range(abortTransactionalIdStart, abortTransactionalIdEnd)
+			.mapToObj(this::generateTransactionalId)
+			.collect(Collectors.toSet());
+	}
+
+	private String generateTransactionalId(long transactionalId) {
+		return String.format(prefix + "-%d", transactionalId);
+	}
+}


Mime
View raw message