[Bug] Master branch is broken due to failing test testElasticSearch8Sink

[eolivelli]

Search before asking

• I searched in the issues and found nothing similar.

Version

master

Minimal reproduce step

Run a Pr against master branch

https://github.com/apache/pulsar/actions/runs/5388118701/jobs/9782385675?pr=20654

What did you expect to see?

The tests are passing

What did you see instead?

The test failed

Anything else?

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!

[eolivelli]

The test is passing locally on my machine on master branch at git sha 2b01f83

[tisonkun]

2b01f83

It's even the latest commit 🤣

Confirmed that I can pass the tests locally with the current master.

[lhotari]

@tisonkun @eolivelli I found a problem in PulsarIOSinkRunner.
Instead of Awaitility, it's using Failsafe.

pulsar/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sinks/PulsarIOSinkRunner.java

Lines 117 to 119 in 46bb63d

	// wait for sink to process messages
	Failsafe.with(statusRetryPolicy).run(() ->
	waitForProcessingSinkMessages(tenant, namespace, sinkName, numMessages));

This is the first exception in the log:

  2023-06-27T11:39:06,756 - INFO  - [main:TestRetrySupport@70] - Detected test failure in test PulsarSinksTest.testElasticSearch8Sink, currentSetupNumber=1
  	at org.testng.Assert.failNotEquals(Assert.java:1577)
  	at org.testng.Assert.assertEqualsImpl(Assert.java:149)
  	at org.testng.Assert.assertEquals(Assert.java:131)
  	at org.testng.Assert.assertEquals(Assert.java:979)
  	at org.testng.Assert.assertEquals(Assert.java:955)
  	at org.testng.Assert.assertEquals(Assert.java:989)
  	at org.apache.pulsar.tests.integration.io.sinks.PulsarIOSinkRunner.waitForProcessingSinkMessages(PulsarIOSinkRunner.java:339)
  	at org.apache.pulsar.tests.integration.io.sinks.PulsarIOSinkRunner.lambda$runSinkTester$4(PulsarIOSinkRunner.java:119)
  	at net.jodah.failsafe.Functions.lambda$toSupplier$10(Functions.java:262)
  	at net.jodah.failsafe.Functions.lambda$get$0(Functions.java:48)
  	at net.jodah.failsafe.RetryPolicyExecutor.lambda$supply$0(RetryPolicyExecutor.java:66)
  	at net.jodah.failsafe.Execution.executeSync(Execution.java:128)
  	at net.jodah.failsafe.FailsafeExecutor.call(FailsafeExecutor.java:379)
  	at net.jodah.failsafe.FailsafeExecutor.run(FailsafeExecutor.java:212)
  	at org.apache.pulsar.tests.integration.io.sinks.PulsarIOSinkRunner.runSinkTester(PulsarIOSinkRunner.java:118)
  	at org.apache.pulsar.tests.integration.io.PulsarIOTestBase.testSink(PulsarIOTestBase.java:40)
  	at org.apache.pulsar.tests.integration.io.sinks.PulsarSinksTest.testElasticSearch8Sink(PulsarSinksTest.java:75)
  	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
  	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  	at java.base/java.lang.reflect.Method.invoke(Method.java:568)
  	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:264)
  	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
  	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
  	at java.base/java.lang.Thread.run(Thread.java:833)

it retries using TestRetrySupport

  2023-06-27T11:39:06,771 - INFO  - [main:TestRetrySupport@53] - Previous test run has failed before PulsarSinksTest.testElasticSearch8Sink, failedSetupNumber=1. Running cleanup and setup.

and fails on the retry:

  java.lang.AssertionError: expected [20] but found [1]
  	at org.testng.Assert.fail(Assert.java:110)
  	at org.testng.Assert.failNotEquals(Assert.java:1577)
  	at org.testng.Assert.assertEqualsImpl(Assert.java:149)
  	at org.testng.Assert.assertEquals(Assert.java:131)
  	at org.testng.Assert.assertEquals(Assert.java:979)
  	at org.testng.Assert.assertEquals(Assert.java:955)
  	at org.testng.Assert.assertEquals(Assert.java:989)
  	at org.apache.pulsar.tests.integration.io.sinks.PulsarIOSinkRunner.waitForProcessingSinkMessages(PulsarIOSinkRunner.java:339)
  	at org.apache.pulsar.tests.integration.io.sinks.PulsarIOSinkRunner.lambda$runSinkTester$4(PulsarIOSinkRunner.java:119)
  	at net.jodah.failsafe.Functions.lambda$toSupplier$10(Functions.java:262)
  	at net.jodah.failsafe.Functions.lambda$get$0(Functions.java:48)
  	at net.jodah.failsafe.RetryPolicyExecutor.lambda$supply$0(RetryPolicyExecutor.java:66)
  	at net.jodah.failsafe.Execution.executeSync(Execution.java:128)
  	at net.jodah.failsafe.FailsafeExecutor.call(FailsafeExecutor.java:379)
  	at net.jodah.failsafe.FailsafeExecutor.run(FailsafeExecutor.java:212)
  	at org.apache.pulsar.tests.integration.io.sinks.PulsarIOSinkRunner.runSinkTester(PulsarIOSinkRunner.java:118)
  	at org.apache.pulsar.tests.integration.io.PulsarIOTestBase.testSink(PulsarIOTestBase.java:40)
  	at org.apache.pulsar.tests.integration.io.sinks.PulsarSinksTest.testElasticSearch8Sink(PulsarSinksTest.java:75)
  	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
  	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  	at java.base/java.lang.reflect.Method.invoke(Method.java:568)
  	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:264)
  	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
  	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
  	at java.base/java.lang.Thread.run(Thread.java:833)

This looks simply because of slow execution and that the retry solution isn't working.

"Retry ... " should get printed by Failsafe when it retries:

pulsar/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/PulsarIOTestRunner.java

Lines 45 to 48 in 82237d3

	protected final RetryPolicy<Void> statusRetryPolicy = new RetryPolicy<Void>()
	.withMaxDuration(ONE_MINUTE)
	.withDelay(TEN_SECONDS)
	.onRetry(e -> log.error("Retry ... "));

That indeed gets printed a few times. I'll take a look if it helps to replace Failsafe with Awaitility.

[tisonkun]

Yeah. I notice that part also. IIRC the last passed run doesn't print "Retry ..."

[lhotari]

btw. I'm running an experiment in my own fork by adding logging to the Elasticsearch container to see what's going on there, lhotari#151 .

[eolivelli]

Thank you @lhotari

Today I cannot continue, can you please pick up this issue ?

[lhotari]

In forks, the Pulsar CI will activate ssh access to the user that triggered the build.

ssh access is restricted to public ssh keys that are listed in the user's github profile.

I'm currently trying to find the reason why the failure happens.

To investigate integration test failures, there's a possibility to leave docker containers running so that problems can be investigated. This makes only sense when triggering the test manually. Here's an example:

# terminate any existing java processes or current running build:
pkill java

# setup Java 17
export JAVA_HOME=/opt/hostedtoolcache/Java_Temurin-Hotspot_jdk/17.0.7-7/x64
export PATH=$JAVA_HOME/bin:$PATH

# leave docker containers running after the test completes
export PULSAR_CONTAINERS_LEAVE_RUNNING=true TESTCONTAINERS_REUSE_ENABLE=true

# run a single test method 
mvn test -DredirectTestOutputToFile=false -DtestRetryCount=0 -Dspotbugs.skip=true -DintegrationTests -pl tests/integration -Dtest=org.apache.pulsar.tests.integration.io.sinks.PulsarSinksTest#testElasticSearch8Sink | tee -a /tmp/test_output.log

# list running containers
docker ps

after investigating, it's possible to kill the pulsar inttest containers

docker kill $(docker ps -q --filter "label=pulsarcontainer=true")

(this is documented in the source code)

deleting containers is required before next run

# delete stopped containers
docker container prune --filter "label=pulsarcontainer=true" -f

[lhotari]

The problem went away during I was investigating it. :(

My current suspicion is the automatic download of MaxMind DB by Elasticsearch. This is a bug reported as elastic/elasticsearch#92335 .

It should be possible to disable downloading by setting .withEnv("ingest.geoip.downloader.enabled", "false"). Let's see if this makes a difference.