[improve][test] Print topic stats and internal-stats if cannot receive message in 30s for Function Test
@@ -39,6 +39,8 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+
+import io.swagger.util.Json;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
@@ -1587,6 +1589,16 @@ public abstract class PulsarFunctionsTest extends PulsarFunctionsTestBase {
for (int i = 0; i < numMessages; i++) {
Message<byte[]> msg = consumer.receive(30, TimeUnit.SECONDS);
+ if (msg == null) {
+ log.info("Input topic stats: {}",
+ Json.pretty(pulsarAdmin.topics().getStats(inputTopic, true)));
+ log.info("Output topic stats: {}",
+ Json.pretty(pulsarAdmin.topics().getStats(outputTopic, true)));
+ log.info("Input topic internal-stats: {}",
+ Json.pretty(pulsarAdmin.topics().getInternalStats(inputTopic, true)));
+ log.info("Output topic internal-stats: {}",
+ Json.pretty(pulsarAdmin.topics().getInternalStats(outputTopic, true)));
+ }
String logMsg = new String(msg.getValue(), UTF_8);
log.info("Received message: '{}'", logMsg);
assertTrue(expectedMessages.contains(logMsg), "Message '" + logMsg + "' not expected");
