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)
