[jira] [Created] (KAFKA-21051) testShareConsumerAfterCoordinatorMovement can halt the test JVM

["Chia-Ping Tsai (Jira)" <ji...@apache.org>]

Chia-Ping Tsai created KAFKA-21051:
--------------------------------------

             Summary: testShareConsumerAfterCoordinatorMovement can halt the test JVM
                 Key: KAFKA-21051
                 URL: https://issues.apache.org/jira/browse/KAFKA-21051
             Project: Kafka
          Issue Type: Test
            Reporter: Chia-Ping Tsai
            Assignee: Chia-Ping Tsai


 
{code:java}
                        if (prodState.done().get() && records.count() == 0) {
                            consState.done().set(true);
                        } {code}
The consumer could end too early due to a rough condition:`records.count()` does not mean there is no more data. Hence, the following assertion fails.

 

 
{code:java}
        // Make sure we consumed all records. Consumed records could be higher
        // due to re-delivery but that is expected since we are only guaranteeing
        // at least once semantics.
        assertTrue(prodState.count().get() <= consState.count().get()); {code}
The failure ends the test without closing the thread which is shutting down the broker.

 

 
{code:java}
                    // shutdown the coordinator
                    KafkaBroker broker = cluster.brokers().get(curShareCoordNodeId.get(0));
                    cluster.shutdownBroker(curShareCoordNodeId.get(0));                    // wait for it to be completely shutdown
                    broker.awaitShutdown(); {code}
As a result, the cleanup of {{ClusterInstance}} skips the broker, but the broker stays alive until the next test.. However, the folder has already been removed, and hence the broker started by the previous test hits a log dir failure, which halts the JVM during the next test.

 
{code:java}
    public void handleLogDirFailure(String dir) {
        LOG.warn("Stopping serving logs in dir {}", dir);
        synchronized (logCreationOrDeletionLock)  {
            liveLogDirs.remove(new File(dir));
            directoryIds.remove(dir);
            if (liveLogDirs.isEmpty()) {
                LOG.error("Shutdown broker because all log dirs in {} have failed", dirsToString(logDirs));
                Exit.halt(1);
            } {code}
 



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