Skip to content

Commit

Permalink
test: fix that some test cases are not running as expected on CI
Browse files Browse the repository at this point in the history
  • Loading branch information
@equanz
equanz committed Nov 10, 2023
1 parent 957337b commit f0b8805
Show file tree
Hide file tree
Showing 10 changed files with 47 additions and 46 deletions.
4 changes: 2 additions & 2 deletions ...mpl/MangedLedgerInterceptorImplTest2.java → ...mpl/MangedLedgerInterceptorImpl2Test.java
View file
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
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
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
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
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
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
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
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
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
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

0 comments on commit f0b8805

Please sign in to comment.