Flaky-test: SystemTopicBasedTopicPoliciesServiceTest.testPrepareInitPoliciesCacheAsyncThrowExceptionAfterCreateReader

[lhotari]

Search before reporting

• I searched in the issues and found nothing similar.

Example failure

https://github.com/apache/pulsar/actions/runs/20263114426/job/58180372254?pr=25079#step:11:991

Exception stacktrace

Error:  Tests run: 24, Failures: 1, Errors: 0, Skipped: 16, Time elapsed: 60.42 s <<< FAILURE! -- in org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesServiceTest
  Error:  org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesServiceTest.testPrepareInitPoliciesCacheAsyncThrowExceptionAfterCreateReader -- Time elapsed: 10.25 s <<< FAILURE!
  org.awaitility.core.ConditionTimeoutException: Assertion condition  within 10 seconds.
  	at org.awaitility.core.ConditionAwaiter.await(ConditionAwaiter.java:167)
  	at org.awaitility.core.AssertionCondition.await(AssertionCondition.java:119)
  	at org.awaitility.core.AssertionCondition.await(AssertionCondition.java:31)
  	at org.awaitility.core.ConditionFactory.until(ConditionFactory.java:985)
  	at org.awaitility.core.ConditionFactory.untilAsserted(ConditionFactory.java:769)
  	at org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesServiceTest.testPrepareInitPoliciesCacheAsyncThrowExceptionAfterCreateReader(SystemTopicBasedTopicPoliciesServiceTest.java:529)
  	at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:103)
  	at java.base/java.lang.reflect.Method.invoke(Method.java:580)
  	at org.testng.internal.invokers.MethodInvocationHelper.invokeMethod(MethodInvocationHelper.java:139)
  	at org.testng.internal.invokers.InvokeMethodRunnable.runOne(InvokeMethodRunnable.java:47)
  	at org.testng.internal.invokers.InvokeMethodRunnable.call(InvokeMethodRunnable.java:76)
  	at org.testng.internal.invokers.InvokeMethodRunnable.call(InvokeMethodRunnable.java:11)
  	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
  	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
  	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
  	at java.base/java.lang.Thread.run(Thread.java:1583)
  Caused by: java.lang.AssertionError: 
  	at org.testng.AssertJUnit.fail(AssertJUnit.java:65)
  	at org.testng.AssertJUnit.assertTrue(AssertJUnit.java:23)
  	at org.testng.AssertJUnit.assertTrue(AssertJUnit.java:33)
  	at org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesServiceTest.lambda$testPrepareInitPoliciesCacheAsyncThrowExceptionAfterCreateReader$30(SystemTopicBasedTopicPoliciesServiceTest.java:532)
  	at org.awaitility.core.AssertionCondition.lambda$new$0(AssertionCondition.java:53)
  	at org.awaitility.core.ConditionAwaiter$ConditionPoller.call(ConditionAwaiter.java:248)
  	at org.awaitility.core.ConditionAwaiter$ConditionPoller.call(ConditionAwaiter.java:235)
  	... 4 more

Are you willing to submit a PR?

• I'm willing to submit a PR!

[lhotari]

@TakaHiR07 Do you have a chance to take a look? This test was added in #24980

[TakaHiR07]

@TakaHiR07 Do you have a chance to take a look? This test was added in #24980

@lhotari
I can't find the root cause of this flaky test. I add some logs and have ran the test in personal ci three times, and all of them pass. You can see here, TakaHiR07#24

After check the code in main logic, I prefer the code logic is ok and it is more likely a problem in test. I suspect the following possibilities:

1. The process is fine, but the logAppender do not output the log information.
2. The logic inside readMorePoliciesAsync() is executed in another single-thread "broker-client-shared-internal-executor", and this logic runs slowly.
3. There are two SystemTopicBasedTopicPoliciesService in test, original one and spy one. Both of them do prepareInitPoliciesCacheAsync() and use the same thread "broker-client-shared-internal-executor". I’m not sure whether it make the process slowly or stuck the process. We can remove the original one in test code, but I wonder whether it is the root cause.

Here are the two SystemTopicBasedTopicPoliciesService class both do readMorePoliciesAsync() in "broker-client-shared-internal-executor" concurrently.

[lhotari]

1. The process is fine, but the logAppender do not output the log information.

pulsar/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java

Lines 484 to 490 in 47b8d5d

	List<String> logMessages = new ArrayList<>();
	AbstractAppender appender = new AbstractAppender("TestAppender", null, null) {
	@Override
	public void append(LogEvent event) {
	logMessages.add(event.getMessage().getFormattedMessage());
	}
	};

@TakaHiR07 This is most likely the root cause. The appender is using an ArrayList to capture the events. That's not thread safe. I'd suggesting switching to use org.apache.pulsar.utils.TestLogAppender for capturing the events. There's a usage example in another test:

pulsar/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MultiTopicsConsumerTest.java

Lines 500 to 514 in 43f75df

	@Cleanup
	TestLogAppender testLogAppender = TestLogAppender.create(Optional.empty());
	consumer.close();
	consumerClosedLatch.countDown();
	// only a single message should be processed
	assertEquals(messageProcessedCount.get(), 1);
	// wait until all messages have been drained from the executor
	Awaitility.await().untilAsserted(() -> assertEquals(messagesCurrentlyInExecutor.get(), 0));
	testLogAppender.getEvents().forEach(logEvent -> {
	if (logEvent.getLevel() == Level.ERROR) {
	org.apache.logging.log4j.message.Message logEventMessage = logEvent.getMessage();
	fail("No error should be logged when closing a consumer. Got: " + logEventMessage
	.getFormattedMessage() + " throwable:" + logEventMessage.getThrowable());
	}
	});

[lhotari]

A quick fix would be to use List<String> logMessages = Collections.synchronizedList(new ArrayList<>());

[TakaHiR07]

@lhotari Have push a pr fix.