[Transaction] Fix initTransaction might wait until request timeout

[BewareMyPower]

Motivation

Most of the tests in TransactionTest take at least 10 seconds because the request.timeout.ms config is 10000. See

kop/tests/src/test/java/io/streamnative/pulsar/handlers/kop/coordinator/transaction/TransactionTest.java

Line 1150 in 1fd3bdb

producerProps.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 1000 * 10);

It's because KoP loads the metadata topic lazily, unlike Kafka, which loads the metadata topic when started. KoP adopts this way because KoP starts before the Pulsar broker finishes the start, in this case there might be some problems with the topic lookup.

Therefore, when the Kafka producer sends the 1st INIT_PRODUCER_ID request, it will receive COORDINATOR_LOAD_IN_PROGRESS error, and the request will reenqueue [1], then the request will expire after request.timeout.ms milliseconds.

} else if (error != Errors.NOT_COORDINATOR && error != Errors.COORDINATOR_NOT_AVAILABLE) {
    if (error != Errors.COORDINATOR_LOAD_IN_PROGRESS && error != Errors.CONCURRENT_TRANSACTIONS) {
       /* ... */
    } else {
        this.reenqueue(); // [1]
    }
} else { // [2]
    TransactionManager.this.lookupCoordinator(CoordinatorType.TRANSACTION, TransactionManager.this.transactionalId);
    this.reenqueue();
}

Modifications

Return COORDINATOR_NOT_AVAILABLE instead of
COORDINATOR_LOAD_IN_PROGRESS when the transaction metadata topic is loading. It's different with Kafka's behavior but it can avoid waiting 30 seconds by default for initTransaction in KoP.

In addition, configure the request timeout with 3 seconds for some cases that initTransaction might expire so that the test time could be reduced.

Documentation

Check the box below.

Need to update docs?

• doc-required

  (If you need help on updating docs, create a doc issue)

• no-need-doc

  (Please explain why)

• doc

  (If this PR contains doc changes)

[BewareMyPower]

The test takes before 2min 57s before this PR:

The test takes 1min 56s after this PR:

[Demogorgon314]

Should we add a comment to let others know why we change to COORDINATOR_NOT_AVAILABLE?

[gaoran10]

Maybe it's also related to this change, I think the init-producerID request should be resent after 1 second when receiving COORDINATOR_LOAD_IN_PROGRESS exception, but it doesn't.

[Demogorgon314]

Maybe it's also related to this change, I think the init-producerID request should be resent after 1 second when receiving COORDINATOR_LOAD_IN_PROGRESS exception, but it doesn't.

Yes, it will wait for the request timeout and then send the next request. I tried to change the ThrottleTimeMs to -1. Then it won't wait for the request timeout. Maybe we should investigate this.

[BewareMyPower]

@gaoran10 Good find! I will investigate it deeper.

[BewareMyPower]

Yes, it's because of the throttle. When any response arrives, the NetworkClient#maybeThrottle is called before passing the response to the callback.

    private void maybeThrottle(AbstractResponse response, short apiVersion, String nodeId, long now) {
        int throttleTimeMs = response.throttleTimeMs();
        if (throttleTimeMs > 0 && response.shouldClientThrottle(apiVersion)) {
            connectionStates.throttle(nodeId, now + throttleTimeMs);
            log.trace("Connection to node {} is throttled for {} ms until timestamp {}", nodeId, throttleTimeMs,
                      now + throttleTimeMs);
        }
    }

However, the throttle is only 1 second. After debugging with Kafka client, I found the root cause is Sender#awaitNodeReady. I added some logs

                log.info("XYZ before awaitNodeReady");
                if (!awaitNodeReady(targetNode, coordinatorType)) {
                    /* ... */
                } else {
                    log.trace("XYZ awaitNodeReady done: {} {}", targetNode, coordinatorType);
                }

and found:

[2023-08-03 16:31:34,781] INFO [Producer clientId=producer-txn-id, transactionalId=txn-id] XYZ before awaitNodeReady (org.apache.kafka.clients.producer.internals.Sender:475)
[2023-08-03 16:31:44,783] TRACE [Producer clientId=producer-txn-id, transactionalId=txn-id] XYZ awaitNodeReady done: localhost:15003 (id: 1825631472 rack: null) TRANSACTION (org.apache.kafka.clients.producer.internals.Sender:481)

NetworkClientUtils#awaitReady, which is called by Sender#awaitNodeReady, is the root cause. See my code comments below.

public static boolean awaitReady(KafkaClient client, Node node, Time time, long timeoutMs) throws IOException {
    // timeoutMs is the request timeout, which is 10 seconds in our config.
    if (timeoutMs < 0) {
        throw new IllegalArgumentException("Timeout needs to be greater than 0");
    }
    long startTime = time.milliseconds();

    // isReady will check if `throttleUntilTimeMs` of the connection is before the current timestamp.
    if (isReady(client, node, startTime) ||  client.ready(node, startTime))
        return true;

    long attemptStartTime = time.milliseconds();
    while (!client.isReady(node, attemptStartTime) && attemptStartTime - startTime < timeoutMs) {
        // Since the throttle time is 1 second, it will always go to this loop 
        if (client.connectionFailed(node)) {
            throw new IOException("Connection to " + node + " failed.");
        }
        // Since the time of `isReady()` and `client.ready()` calls are too small to count,
        // the pollTimeout is very closed to the `timeoutMs` (10 seconds)
        long pollTimeout = timeoutMs - (attemptStartTime - startTime);
        // Then, `poll` will be blocking for nearly 10 seconds because there is no inflight request
        client.poll(pollTimeout, attemptStartTime);
        if (client.authenticationException(node) != null)
            throw client.authenticationException(node);
        attemptStartTime = time.milliseconds();
    }
    return client.isReady(node, attemptStartTime);
}

The reason why returning a COORDINATOR_NOT_AVAILABLE is that before reenqueue, a FIND_COORDINATOR request is sent. In this case, maybeSendAndPollTransactionalRequest calls maybeFindCoordinatorAndRetry directly:

            } else if (coordinatorType != null) {
                log.trace("Coordinator not known for {}, will retry {} after finding coordinator.", coordinatorType, requestBuilder.apiKey());
                maybeFindCoordinatorAndRetry(nextRequestHandler);
                return true;

[codecov]

Codecov Report

Merging #1991 (227876c) into master (0241e43) will increase coverage by 0.00%.
The diff coverage is n/a.

@@            Coverage Diff            @@
##             master    #1991   +/-   ##
=========================================
  Coverage     17.29%   17.29%           
  Complexity      728      728           
=========================================
  Files           190      190           
  Lines         14041    14038    -3     
  Branches       1320     1318    -2     
=========================================
  Hits           2428     2428           
+ Misses        11437    11434    -3     
  Partials        176      176           

Files Changed	Coverage Δ
...ative/pulsar/handlers/kop/KafkaRequestHandler.java	1.06% <ø> (+<0.01%)	⬆️

[BewareMyPower]

Hi @Demogorgon314 @gaoran10, I just updated my PR according to the explanation above. IMO, it does not make sense to set a 1 second throttle ms in the InitProducerId response, so I removed the config.