[jira] [Commented] (FLINK-37356) Recycle use of kafka producer(which commit error) maybe send data without AddPartitionsToTxnRequest

["Arvid Heise (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-37356?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17928691#comment-17928691 ] 

Arvid Heise commented on FLINK-37356:
-------------------------------------

Thank you very much for the detailed analysis. However, I'm still not sure I fully understand the issue.

In general, everything that we do with reflection is fishy and may have worked at some point in time but prone to fail at a later time. So that makes your analysis very plausible.

What I didn't fully understand is: how do we get an entry in {{partitionsInTransaction}} when we have an empty data? As far as I understood FLINK-31363, we actually don't have any partitions and as such the transaction is invisible to the broker.

Nevertheless, I do see two issues:
* Committing an empty transaction will result in a {{{}InvalidTxnStateException{}}}. I think that the current {{main}} contains already a fix for that by simple not trying to commit and resetting the state only on client side (please double-check if that makes sense at all).
* I also think it may be inherently unsafe to reuse a producer that encountered an error for any reason. Let's say the producer the producer was fenced. Can we assume that the transaction manager is in a state from which we can easily resume? In theory, {{initTransaction}} should reset everything but you definitively found cases where it doesn't. So I'd propose to close a producer that encountered a non-retriable error and open a new one instead.

Coming back to your original problem. It would be awesome if you could distill the issue to a small repo/test case that fails. From your description it sounds easy to do, although - as I said - I cannot follow 100%. {{main}} now also contains better debug logging that makes it easier to follow reuse because I also print the system hash code. That also helps tremendously in debugging leaks.

> Recycle use of kafka producer(which commit error) maybe send data without AddPartitionsToTxnRequest
> ---------------------------------------------------------------------------------------------------
>
>                 Key: FLINK-37356
>                 URL: https://issues.apache.org/jira/browse/FLINK-37356
>             Project: Flink
>          Issue Type: Improvement
>          Components: Connectors / Kafka
>    Affects Versions: kafka-3.4.0
>            Reporter: Hongshun Wang
>            Priority: Major
>
> In my production environment,  READ_COMMITTED consumer can no longer consume any records. Then I found that the LSO of the partition doesn't change for a long time.  I lookup all the log in Kafka cluster, then find that there is a transaction lacking AddPartitionsToTxnRequest.
>  
> At first, I think the problem is caused by https://issues.apache.org/jira/browse/FLINK-31363 because my Kafka cluster log contains InvalidTxnStateException. However, though the transaction is in an invalid state,  no data is written into Kafka topic partition in this transaction( because in this case, the transaction is empty). It will not influence any Kafka topic partition's LSO, thus consumer won't be blocked.
>  
> Then I check the code of Kafka client, it seems no way to produce data without AddPartitionsToTxnRequest done because the the `Sender` will refuse to dequeue batches from the accumulator until they have been added to the transaction.
> {code:java}
> // org.apache.kafka.clients.producer.KafkaProducer#doSend
> private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) {
> // ..ignore code
> // Add the partition to the transaction (if in progress) after it has been successfully
> // appended to the accumulator. We cannot do it before because the partition may be
> // unknown or the initially selected partition may be changed when the batch is closed
> // (as indicated by `abortForNewBatch`). Note that the `Sender` will refuse to dequeue
> // batches from the accumulator until they have been added to the transaction.
> if (transactionManager != null) {
>     transactionManager.maybeAddPartition(appendCallbacks.topicPartition());
> // ignore code
> }{code}
> {code:java}
> //org.apache.kafka.clients.producer.internals.RecordAccumulator#drainBatchesForOneNode
>  
> if (size + first.estimatedSizeInBytes() > maxSize && !ready.isEmpty()) { // there is a rare case that a single batch size is larger than the request size due to // compression; in this case we will still eventually send this batch in a single request break; } else { if (shouldStopDrainBatchesForPartition(first, tp)) break; }
> {code}
>  
> Then I have a idea that if a TransactionManager which doesn't clear 
> partitionsInTransaction is reused again, the AddPartitionsToTxnRequest will be sent again.  It maybe happen in Flink Kafka connector:
>  
>       1. The flink kafka connector also reuse and recycle KafkaProducer: 
> KafkaCommitter will recycle the producer to
> producerPool after the transaction complete or exception,  and then KafkaWriter will reuse it from producerPool.
> {code:java}
> // code placeholder
> org.apache.flink.connector.kafka.sink.KafkaCommitter#commit
> @Override
> public void commit(Collection<CommitRequest<KafkaCommittable>> requests)
>         throws IOException, InterruptedException {
>     for (CommitRequest<KafkaCommittable> request : requests) {
>              
>             producer.commitTransaction();
>             producer.flush();
>             recyclable.ifPresent(Recyclable::close);
>         } catch (RetriableException e) {
>             request.retryLater();
>         } catch (ProducerFencedException e) {
>             recyclable.ifPresent(Recyclable::close);
>             request.signalFailedWithKnownReason(e);
>         } catch (InvalidTxnStateException e) {
>             recyclable.ifPresent(Recyclable::close);
>             request.signalFailedWithKnownReason(e);
>         } catch (UnknownProducerIdException e) {
>             LOG.error(
>             recyclable.ifPresent(Recyclable::close);
>             request.signalFailedWithKnownReason(e);
>         } catch (Exception e) {
>             recyclable.ifPresent(Recyclable::close);
>             request.signalFailedWithUnknownReason(e);
>         }
>     }
> } {code}
>      2. If KafkaCommitter meet an exception and doesn't sucess to commitTransaction, the partitionsInTransaction in 
> TransactionManager won't be clear(org.apache.kafka.clients.producer.internals.TransactionManager#completeTransaction). 
>    3. If KafkaWriter which reuse same producer and send data to  same partitions in next transaction, the AddPartitionsToTxnRequest won't be send.
>  
> Thus, in FlinkKafkaInternalProducer#setTransactionId, we should clear the partition information of TransactionManager.(now we just set 
> transactionalId and currentState.
> {code:java}
> // code placeholder
> public void setTransactionId(String transactionalId) {
>     if (!transactionalId.equals(this.transactionalId)) {
>         checkState(
>                 !inTransaction,
>                 String.format("Another transaction %s is still open.", transactionalId));
>         LOG.debug("Change transaction id from {} to {}", this.transactionalId, transactionalId);
>         Object transactionManager = getTransactionManager();
>         synchronized (transactionManager) {
>             setField(transactionManager, "transactionalId", transactionalId);
>             setField(
>                     transactionManager,
>                     "currentState",
>                     getTransactionManagerState("UNINITIALIZED"));
>             this.transactionalId = transactionalId;
>         }
>     }
> } {code}
>  
>  
>  
>  



--
This message was sent by Atlassian Jira
(v8.20.10#820010)