Skip to content

[fix][test] Fix that some test cases are not running as expected on CI #21532

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lhotari merged 1 commit into from
Nov 10, 2023
Merged

[fix][test] Fix that some test cases are not running as expected on CI #21532

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions ...mpl/MangedLedgerInterceptorImplTest2.java → ...mpl/MangedLedgerInterceptorImpl2Test.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,9 +39,9 @@
*/
@Slf4j
@Test(groups = "broker")
public class MangedLedgerInterceptorImplTest2 extends MockedBookKeeperTestCase {
public class MangedLedgerInterceptorImpl2Test extends MockedBookKeeperTestCase {

public static void switchLedgerManually(ManagedLedgerImpl ledger){
private static void switchLedgerManually(ManagedLedgerImpl ledger){
LedgerHandle originalLedgerHandle = ledger.currentLedger;
ledger.ledgerClosed(ledger.currentLedger);
ledger.createLedgerAfterClosed();
Expand Down
34 changes: 17 additions & 17 deletions ...r-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionLocalRunTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -538,15 +538,15 @@ private void testE2EPulsarFunctionLocalRun(String jarFilePathUrl, int parallelis
totalMsgs);

// validate prometheus metrics
String prometheusMetrics = PulsarFunctionTestUtils.getPrometheusMetrics(metricsPort);
String prometheusMetrics = TestPulsarFunctionUtils.getPrometheusMetrics(metricsPort);
log.info("prometheus metrics: {}", prometheusMetrics);

Map<String, PulsarFunctionTestUtils.Metric> metricsMap = new HashMap<>();
Map<String, TestPulsarFunctionUtils.Metric> metricsMap = new HashMap<>();
Arrays.asList(prometheusMetrics.split("\n")).forEach(line -> {
if (line.startsWith("pulsar_function_processed_successfully_total")) {
Map<String, PulsarFunctionTestUtils.Metric> metrics = PulsarFunctionTestUtils.parseMetrics(line);
Map<String, TestPulsarFunctionUtils.Metric> metrics = TestPulsarFunctionUtils.parseMetrics(line);
assertFalse(metrics.isEmpty());
PulsarFunctionTestUtils.Metric m = metrics.get("pulsar_function_processed_successfully_total");
TestPulsarFunctionUtils.Metric m = metrics.get("pulsar_function_processed_successfully_total");
if (m != null) {
metricsMap.put(m.tags.get("instance_id"), m);
}
Expand All @@ -556,7 +556,7 @@ private void testE2EPulsarFunctionLocalRun(String jarFilePathUrl, int parallelis

double totalMsgRecv = 0.0;
for (int i = 0; i < parallelism; i++) {
PulsarFunctionTestUtils.Metric m = metricsMap.get(String.valueOf(i));
TestPulsarFunctionUtils.Metric m = metricsMap.get(String.valueOf(i));
Assert.assertNotNull(m);
assertEquals(m.tags.get("cluster"), config.getClusterName());
assertEquals(m.tags.get("instance_id"), String.valueOf(i));
Expand Down Expand Up @@ -843,15 +843,15 @@ private void testPulsarSourceLocalRun(String jarFilePathUrl, int parallelism) th
assertEquals(admin.topics().getStats(sinkTopic).getPublishers().size(), parallelism);

// validate prometheus metrics
String prometheusMetrics = PulsarFunctionTestUtils.getPrometheusMetrics(metricsPort);
String prometheusMetrics = TestPulsarFunctionUtils.getPrometheusMetrics(metricsPort);
log.info("prometheus metrics: {}", prometheusMetrics);

Map<String, PulsarFunctionTestUtils.Metric> metricsMap = new HashMap<>();
Map<String, TestPulsarFunctionUtils.Metric> metricsMap = new HashMap<>();
Arrays.asList(prometheusMetrics.split("\n")).forEach(line -> {
if (line.startsWith("pulsar_source_written_total")) {
Map<String, PulsarFunctionTestUtils.Metric> metrics = PulsarFunctionTestUtils.parseMetrics(line);
Map<String, TestPulsarFunctionUtils.Metric> metrics = TestPulsarFunctionUtils.parseMetrics(line);
assertFalse(metrics.isEmpty());
PulsarFunctionTestUtils.Metric m = metrics.get("pulsar_source_written_total");
TestPulsarFunctionUtils.Metric m = metrics.get("pulsar_source_written_total");
if (m != null) {
metricsMap.put(m.tags.get("instance_id"), m);
}
Expand All @@ -860,7 +860,7 @@ private void testPulsarSourceLocalRun(String jarFilePathUrl, int parallelism) th
Assert.assertEquals(metricsMap.size(), parallelism);

for (int i = 0; i < parallelism; i++) {
PulsarFunctionTestUtils.Metric m = metricsMap.get(String.valueOf(i));
TestPulsarFunctionUtils.Metric m = metricsMap.get(String.valueOf(i));
Assert.assertNotNull(m);
assertEquals(m.tags.get("cluster"), config.getClusterName());
assertEquals(m.tags.get("instance_id"), String.valueOf(i));
Expand Down Expand Up @@ -1002,22 +1002,22 @@ private void testPulsarSinkLocalRun(String jarFilePathUrl, int parallelism, Stri
}, 5, 200));

// validate prometheus metrics
String prometheusMetrics = PulsarFunctionTestUtils.getPrometheusMetrics(metricsPort);
String prometheusMetrics = TestPulsarFunctionUtils.getPrometheusMetrics(metricsPort);
log.info("prometheus metrics: {}", prometheusMetrics);

Map<String, PulsarFunctionTestUtils.Metric> metricsMap = new HashMap<>();
Map<String, TestPulsarFunctionUtils.Metric> metricsMap = new HashMap<>();
Arrays.asList(prometheusMetrics.split("\n")).forEach(line -> {
if (line.startsWith("pulsar_sink_written_total")) {
Map<String, PulsarFunctionTestUtils.Metric> metrics = PulsarFunctionTestUtils.parseMetrics(line);
Map<String, TestPulsarFunctionUtils.Metric> metrics = TestPulsarFunctionUtils.parseMetrics(line);
assertFalse(metrics.isEmpty());
PulsarFunctionTestUtils.Metric m = metrics.get("pulsar_sink_written_total");
TestPulsarFunctionUtils.Metric m = metrics.get("pulsar_sink_written_total");
if (m != null) {
metricsMap.put(m.tags.get("instance_id"), m);
}
} else if (line.startsWith("pulsar_sink_sink_exceptions_total")) {
Map<String, PulsarFunctionTestUtils.Metric> metrics = PulsarFunctionTestUtils.parseMetrics(line);
Map<String, TestPulsarFunctionUtils.Metric> metrics = TestPulsarFunctionUtils.parseMetrics(line);
assertFalse(metrics.isEmpty());
PulsarFunctionTestUtils.Metric m = metrics.get("pulsar_sink_sink_exceptions_total");
TestPulsarFunctionUtils.Metric m = metrics.get("pulsar_sink_sink_exceptions_total");
if (m == null) {
m = metrics.get("pulsar_sink_sink_exceptions_1min_total");
}
Expand All @@ -1028,7 +1028,7 @@ private void testPulsarSinkLocalRun(String jarFilePathUrl, int parallelism, Stri

double totalNumRecvMsg = 0;
for (int i = 0; i < parallelism; i++) {
PulsarFunctionTestUtils.Metric m = metricsMap.get(String.valueOf(i));
TestPulsarFunctionUtils.Metric m = metricsMap.get(String.valueOf(i));
Assert.assertNotNull(m);
assertEquals(m.tags.get("cluster"), config.getClusterName());
assertEquals(m.tags.get("instance_id"), String.valueOf(i));
Expand Down
3 changes: 2 additions & 1 deletion ...tions/worker/PulsarFunctionTestUtils.java → ...tions/worker/TestPulsarFunctionUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,8 @@
import static com.google.common.base.Preconditions.checkArgument;

@Slf4j
public class PulsarFunctionTestUtils {
@Test(groups = "functions-worker")
public class TestPulsarFunctionUtils {
public static String getPrometheusMetrics(int metricsPort) throws IOException {
StringBuilder result = new StringBuilder();
URL url = new URL(String.format("http://%s:%s/metrics", "localhost", metricsPort));
Expand Down
8 changes: 4 additions & 4 deletions pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarBatchSourceE2ETest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@
import org.apache.pulsar.common.policies.data.TopicStats;
import org.apache.pulsar.functions.utils.FunctionCommon;

import org.apache.pulsar.functions.worker.PulsarFunctionTestUtils;
import org.apache.pulsar.functions.worker.TestPulsarFunctionUtils;
import org.apache.pulsar.io.batchdiscovery.ImmediateTriggerer;
import org.testng.annotations.Test;

Expand Down Expand Up @@ -102,11 +102,11 @@ private void testPulsarBatchSourceStats(String jarFilePathUrl) throws Exception
}, 50, 150);
assertEquals(admin.topics().getStats(sinkTopic2).getPublishers().size(), 1);

String prometheusMetrics = PulsarFunctionTestUtils.getPrometheusMetrics(pulsar.getListenPortHTTP().get());
String prometheusMetrics = TestPulsarFunctionUtils.getPrometheusMetrics(pulsar.getListenPortHTTP().get());
log.info("prometheusMetrics: {}", prometheusMetrics);

Map<String, PulsarFunctionTestUtils.Metric> metrics = PulsarFunctionTestUtils.parseMetrics(prometheusMetrics);
PulsarFunctionTestUtils.Metric m = metrics.get("pulsar_source_received_total");
Map<String, TestPulsarFunctionUtils.Metric> metrics = TestPulsarFunctionUtils.parseMetrics(prometheusMetrics);
TestPulsarFunctionUtils.Metric m = metrics.get("pulsar_source_received_total");
assertEquals(m.tags.get("cluster"), config.getClusterName());
assertEquals(m.tags.get("instance_id"), "0");
assertEquals(m.tags.get("name"), sourceName);
Expand Down
12 changes: 6 additions & 6 deletions pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarFunctionE2ETest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,7 +67,7 @@
import org.apache.pulsar.functions.instance.InstanceUtils;
import org.apache.pulsar.functions.utils.FunctionCommon;
import org.apache.pulsar.functions.worker.FunctionRuntimeManager;
import org.apache.pulsar.functions.worker.PulsarFunctionTestUtils;
import org.apache.pulsar.functions.worker.TestPulsarFunctionUtils;
import org.awaitility.Awaitility;
import org.testng.Assert;
import org.testng.annotations.Test;
Expand Down Expand Up @@ -373,11 +373,11 @@ public void testPulsarFunctionStats() throws Exception {
functionStats.getAvgProcessLatency());

// validate prometheus metrics empty
String prometheusMetrics = PulsarFunctionTestUtils.getPrometheusMetrics(pulsar.getListenPortHTTP().get());
String prometheusMetrics = TestPulsarFunctionUtils.getPrometheusMetrics(pulsar.getListenPortHTTP().get());
log.info("prometheus metrics: {}", prometheusMetrics);

Map<String, PulsarFunctionTestUtils.Metric> metrics = PulsarFunctionTestUtils.parseMetrics(prometheusMetrics);
PulsarFunctionTestUtils.Metric m = metrics.get("pulsar_function_received_total");
Map<String, TestPulsarFunctionUtils.Metric> metrics = TestPulsarFunctionUtils.parseMetrics(prometheusMetrics);
TestPulsarFunctionUtils.Metric m = metrics.get("pulsar_function_received_total");
assertEquals(m.tags.get("cluster"), config.getClusterName());
assertEquals(m.tags.get("instance_id"), "0");
assertEquals(m.tags.get("name"), functionName);
Expand Down Expand Up @@ -533,10 +533,10 @@ public void testPulsarFunctionStats() throws Exception {
assertEquals(functionInstanceStats, functionStats.instances.get(0).getMetrics());

// validate prometheus metrics
prometheusMetrics = PulsarFunctionTestUtils.getPrometheusMetrics(pulsar.getListenPortHTTP().get());
prometheusMetrics = TestPulsarFunctionUtils.getPrometheusMetrics(pulsar.getListenPortHTTP().get());
log.info("prometheus metrics: {}", prometheusMetrics);

metrics = PulsarFunctionTestUtils.parseMetrics(prometheusMetrics);
metrics = TestPulsarFunctionUtils.parseMetrics(prometheusMetrics);
m = metrics.get("pulsar_function_received_total");
assertEquals(m.tags.get("cluster"), config.getClusterName());
assertEquals(m.tags.get("instance_id"), "0");
Expand Down
18 changes: 9 additions & 9 deletions pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSinkE2ETest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,7 @@
import org.apache.pulsar.functions.LocalRunner;
import org.apache.pulsar.functions.api.Record;
import org.apache.pulsar.functions.utils.FunctionCommon;
import org.apache.pulsar.functions.worker.PulsarFunctionTestUtils;
import org.apache.pulsar.functions.worker.TestPulsarFunctionUtils;
import org.apache.pulsar.io.core.Sink;
import org.apache.pulsar.io.core.SinkContext;
import org.awaitility.Awaitility;
Expand Down Expand Up @@ -122,9 +122,9 @@ public void testReadCompactedSink() throws Exception {

// 5 Sink should only read compacted value, so we will only receive compacted messages
Awaitility.await().ignoreExceptions().untilAsserted(() -> {
String prometheusMetrics = PulsarFunctionTestUtils.getPrometheusMetrics(pulsar.getListenPortHTTP().get());
Map<String, PulsarFunctionTestUtils.Metric> metrics = PulsarFunctionTestUtils.parseMetrics(prometheusMetrics);
PulsarFunctionTestUtils.Metric m = metrics.get("pulsar_sink_received_total");
String prometheusMetrics = TestPulsarFunctionUtils.getPrometheusMetrics(pulsar.getListenPortHTTP().get());
Map<String, TestPulsarFunctionUtils.Metric> metrics = TestPulsarFunctionUtils.parseMetrics(prometheusMetrics);
TestPulsarFunctionUtils.Metric m = metrics.get("pulsar_sink_received_total");
assertEquals(m.value, maxKeys);
});
}
Expand Down Expand Up @@ -271,11 +271,11 @@ private void testPulsarSinkStats(String jarFilePathUrl, Function<SinkConfig, Sin
status.getInstances().forEach(sinkInstanceStatus -> assertEquals(sinkInstanceStatus.status.numSystemExceptions, 0));

// validate prometheus metrics empty
String prometheusMetrics = PulsarFunctionTestUtils.getPrometheusMetrics(pulsar.getListenPortHTTP().get());
String prometheusMetrics = TestPulsarFunctionUtils.getPrometheusMetrics(pulsar.getListenPortHTTP().get());
log.info("prometheus metrics: {}", prometheusMetrics);

Map<String, PulsarFunctionTestUtils.Metric> metrics = PulsarFunctionTestUtils.parseMetrics(prometheusMetrics);
PulsarFunctionTestUtils.Metric m = metrics.get("pulsar_sink_received_total");
Map<String, TestPulsarFunctionUtils.Metric> metrics = TestPulsarFunctionUtils.parseMetrics(prometheusMetrics);
TestPulsarFunctionUtils.Metric m = metrics.get("pulsar_sink_received_total");
assertEquals(m.tags.get("cluster"), config.getClusterName());
assertEquals(m.tags.get("instance_id"), "0");
assertEquals(m.tags.get("name"), sinkName);
Expand Down Expand Up @@ -364,10 +364,10 @@ private void testPulsarSinkStats(String jarFilePathUrl, Function<SinkConfig, Sin
status.getInstances().forEach(sinkInstanceStatus -> assertEquals(sinkInstanceStatus.status.numSystemExceptions, 0));

// get stats after producing
prometheusMetrics = PulsarFunctionTestUtils.getPrometheusMetrics(pulsar.getListenPortHTTP().get());
prometheusMetrics = TestPulsarFunctionUtils.getPrometheusMetrics(pulsar.getListenPortHTTP().get());
log.info("prometheusMetrics: {}", prometheusMetrics);

metrics = PulsarFunctionTestUtils.parseMetrics(prometheusMetrics);
metrics = TestPulsarFunctionUtils.parseMetrics(prometheusMetrics);
m = metrics.get("pulsar_sink_received_total");
assertEquals(m.tags.get("cluster"), config.getClusterName());
assertEquals(m.tags.get("instance_id"), "0");
Expand Down
8 changes: 4 additions & 4 deletions pulsar-broker/src/test/java/org/apache/pulsar/io/PulsarSourceE2ETest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@
import org.apache.pulsar.common.io.SourceConfig;
import org.apache.pulsar.common.policies.data.TopicStats;
import org.apache.pulsar.functions.utils.FunctionCommon;
import org.apache.pulsar.functions.worker.PulsarFunctionTestUtils;
import org.apache.pulsar.functions.worker.TestPulsarFunctionUtils;
import org.testng.annotations.Test;

import com.google.common.collect.Lists;
Expand Down Expand Up @@ -106,11 +106,11 @@ private void testPulsarSourceStats(String jarFilePathUrl) throws Exception {
}, 50, 150);
assertEquals(admin.topics().getStats(sinkTopic2).getPublishers().size(), 1);

String prometheusMetrics = PulsarFunctionTestUtils.getPrometheusMetrics(pulsar.getListenPortHTTP().get());
String prometheusMetrics = TestPulsarFunctionUtils.getPrometheusMetrics(pulsar.getListenPortHTTP().get());
log.info("prometheusMetrics: {}", prometheusMetrics);

Map<String, PulsarFunctionTestUtils.Metric> metrics = PulsarFunctionTestUtils.parseMetrics(prometheusMetrics);
PulsarFunctionTestUtils.Metric m = metrics.get("pulsar_source_received_total");
Map<String, TestPulsarFunctionUtils.Metric> metrics = TestPulsarFunctionUtils.parseMetrics(prometheusMetrics);
TestPulsarFunctionUtils.Metric m = metrics.get("pulsar_source_received_total");
assertEquals(m.tags.get("cluster"), config.getClusterName());
assertEquals(m.tags.get("instance_id"), "0");
assertEquals(m.tags.get("name"), sourceName);
Expand Down
2 changes: 1 addition & 1 deletion .../server/ProxyKeyStoreTlsTestWithAuth.java → .../server/ProxyKeyStoreTlsWithAuthTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@
import org.testng.annotations.Test;

@Slf4j
public class ProxyKeyStoreTlsTestWithAuth extends MockedPulsarServiceBaseTest {
public class ProxyKeyStoreTlsWithAuthTest extends MockedPulsarServiceBaseTest {
private ProxyService proxyService;
private ProxyConfiguration proxyConfig = new ProxyConfiguration();

Expand Down
2 changes: 1 addition & 1 deletion ...rver/ProxyKeyStoreTlsTestWithoutAuth.java → ...rver/ProxyKeyStoreTlsWithoutAuthTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,7 @@
import org.testng.annotations.Test;

@Slf4j
public class ProxyKeyStoreTlsTestWithoutAuth extends MockedPulsarServiceBaseTest {
public class ProxyKeyStoreTlsWithoutAuthTest extends MockedPulsarServiceBaseTest {
private ProxyService proxyService;
private ProxyConfiguration proxyConfig = new ProxyConfiguration();

Expand Down
2 changes: 1 addition & 1 deletion ...ar/proxy/server/ProxyTlsTestWithAuth.java → ...ar/proxy/server/ProxyTlsWithAuthTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;

public class ProxyTlsTestWithAuth extends MockedPulsarServiceBaseTest {
public class ProxyTlsWithAuthTest extends MockedPulsarServiceBaseTest {

private ProxyService proxyService;
private ProxyConfiguration proxyConfig = new ProxyConfiguration();
Expand Down