[hotfix][kafka] Do not return producers to a pool in abort for non EXACTLY_ONCE modes

Previously on abort(...) producers were returned to the pool. This was minor bug,
probably without any negative side effect, however this patch fixes it
and adds additional sanity checks to guard against similar bugs
in the future.

This closes #4915.

flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java

@@ -45,7 +45,6 @@ import org.apache.flink.streaming.util.serialization.KeyedSerializationSchemaWra
 import org.apache.flink.streaming.util.serialization.SerializationSchema;
 import org.apache.flink.util.ExceptionUtils;
 import org.apache.flink.util.NetUtils;
-import org.apache.flink.util.Preconditions;
 
 import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
 
@@ -226,7 +225,7 @@ public class FlinkKafkaProducer011<IN>
 	/**
 	 * Pool of KafkaProducers objects.
 	 */
-	private transient ProducersPool producersPool = new ProducersPool();
+	private transient Optional<ProducersPool> producersPool = Optional.empty();
 
 	/**
 	 * Flag controlling whether we are writing the Flink record's timestamp into Kafka.
@@ -596,7 +595,7 @@ public class FlinkKafkaProducer011<IN>
 			asyncException = ExceptionUtils.firstOrSuppressed(e, asyncException);
 		}
 		try {
-			producersPool.close();
+			producersPool.ifPresent(pool -> pool.close());
 		}
 		catch (Exception e) {
 			asyncException = ExceptionUtils.firstOrSuppressed(e, asyncException);
@@ -628,7 +627,7 @@ public class FlinkKafkaProducer011<IN>
 	}
 
 	private FlinkKafkaProducer<byte[], byte[]> createOrGetProducerFromPool() throws Exception {
-		FlinkKafkaProducer<byte[], byte[]> producer = producersPool.poll();
+		FlinkKafkaProducer<byte[], byte[]> producer = getProducersPool().poll();
 		if (producer == null) {
 			String transactionalId = availableTransactionalIds.poll();
 			if (transactionalId == null) {
@@ -661,7 +660,7 @@ public class FlinkKafkaProducer011<IN>
 		switch (semantic) {
 			case EXACTLY_ONCE:
 				transaction.producer.commitTransaction();
-				producersPool.add(transaction.producer);
+				getProducersPool().add(transaction.producer);
 				break;
 			case AT_LEAST_ONCE:
 			case NONE:
@@ -703,11 +702,10 @@ public class FlinkKafkaProducer011<IN>
 		switch (semantic) {
 			case EXACTLY_ONCE:
 				transaction.producer.abortTransaction();
-				producersPool.add(transaction.producer);
+				getProducersPool().add(transaction.producer);
 				break;
 			case AT_LEAST_ONCE:
 			case NONE:
-				producersPool.add(transaction.producer);
 				break;
 			default:
 				throw new UnsupportedOperationException("Not implemented semantic");
@@ -760,7 +758,8 @@ public class FlinkKafkaProducer011<IN>
 		nextTransactionalIdHintState.clear();
 		// To avoid duplication only first subtask keeps track of next transactional id hint. Otherwise all of the
 		// subtasks would write exactly same information.
-		if (getRuntimeContext().getIndexOfThisSubtask() == 0 && nextTransactionalIdHint != null) {
+		if (getRuntimeContext().getIndexOfThisSubtask() == 0 && semantic == Semantic.EXACTLY_ONCE) {
+			checkState(nextTransactionalIdHint != null, "nextTransactionalIdHint must be set for EXACTLY_ONCE");
 			long nextFreeTransactionalId = nextTransactionalIdHint.nextFreeTransactionalId;
 
 			// If we scaled up, some (unknown) subtask must have created new transactional ids from scratch. In that
@@ -788,7 +787,10 @@ public class FlinkKafkaProducer011<IN>
 
 		if (semantic != Semantic.EXACTLY_ONCE) {
 			nextTransactionalIdHint = null;
+			producersPool = Optional.empty();
 		} else {
+			producersPool = Optional.of(new ProducersPool());
+
 			ArrayList<NextTransactionalIdHint> transactionalIdHints = Lists.newArrayList(nextTransactionalIdHintState.get());
 			if (transactionalIdHints.size() > 1) {
 				throw new IllegalStateException(
@@ -829,8 +831,7 @@ public class FlinkKafkaProducer011<IN>
 	}
 
 	private Set<String> generateNewTransactionalIds() {
-		Preconditions.checkState(nextTransactionalIdHint != null,
-			"nextTransactionalIdHint must be present for EXACTLY_ONCE");
+		checkState(nextTransactionalIdHint != null, "nextTransactionalIdHint must be present for EXACTLY_ONCE");
 
 		// range of available transactional ids is:
 		// [nextFreeTransactionalId, nextFreeTransactionalId + parallelism * kafkaProducersPoolSize)
@@ -903,6 +904,11 @@ public class FlinkKafkaProducer011<IN>
 		return initProducer(registerMetrics);
 	}
 
+	private ProducersPool getProducersPool() {
+		checkState(producersPool.isPresent(), "Trying to access uninitialized producer pool");
+		return producersPool.get();
+	}
+
 	private FlinkKafkaProducer<byte[], byte[]> initProducer(boolean registerMetrics) {
 		FlinkKafkaProducer<byte[], byte[]> producer = new FlinkKafkaProducer<>(this.producerConfig);
 
@@ -958,7 +964,7 @@ public class FlinkKafkaProducer011<IN>
 
 	private void readObject(java.io.ObjectInputStream in) throws IOException, ClassNotFoundException {
 		in.defaultReadObject();
-		producersPool = new ProducersPool();
+		producersPool = Optional.empty();
 	}
 
 	private static Properties getPropertiesFromBrokerList(String brokerList) {