pnowojski commented on a change in pull request #9287: [FLINK-13498][kafka]
abort transactions in parallel
URL: https://github.com/apache/flink/pull/9287#discussion_r309072724
##########
File path:
flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011.java
##########
@@ -906,11 +907,33 @@ private void
resetAvailableTransactionalIdsPool(Collection<String> transactional
// ----------------------------------- Utilities
--------------------------
private void abortTransactions(Set<String> transactionalIds) {
- for (String transactionalId : transactionalIds) {
- try (FlinkKafkaProducer<byte[], byte[]> kafkaProducer =
-
initTransactionalProducer(transactionalId, false)) {
- // it suffice to call initTransactions - this
will abort any lingering transactions
- kafkaProducer.initTransactions();
+ // shortcut for non-exactly-once producers
+ if (transactionalIds.isEmpty()) {
+ return;
+ }
+
+ ForkJoinPool forkJoinPool = null;
+ try {
+ // limit the number of connections to the number that
is used during runtime
Review comment:
One more thought - what if kafka/zookeeper are limiting number of concurrent
connections/operations? Technically speaking yes, previously we COULD have the
same number of active `KafkaProducer`s during runtime but:
1. **usually**, only 2 of them were being used - here it's always
`kafkaProducersPoolSize`. This argument maybe we can ignore, since we can argue
that the user's setup should always support `kafkaProducersPoolSize` number of
active `KafkaProducer`s.
2. however, previously `KafkaProducers` they were not being used
concurrently. If we use them concurrently now, we can briefly flood kafka
brokers. Can this be an issue?
Can you @NicoK maybe clear this out with @becketqin? He has better
understanding of how Kafka internally works.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
