From d954feea166c78a0472a51b4b5e3b0435d373aa7 Mon Sep 17 00:00:00 2001 From: lixiaobao <977734161@qq.com> Date: Fri, 11 Apr 2025 15:25:10 +0800 Subject: [PATCH] [Bug] When using SubscriptionPullConsumer to consume data, an error is reported when configuring ConsumerConstant.NODE_URLS_KEY as the cluster address. code: Properties consumerConfig = new Properties(); consumerConfig.put(ConsumerConstant.CONSUMER_ID_KEY, "c1"); consumerConfig.put(ConsumerConstant.CONSUMER_GROUP_ID_KEY, "cg1"); consumerConfig.put(ConsumerConstant.NODE_URLS_KEY, Arrays.asList("192.168.1.1:6667","192.168.1.2:6667","192.168.1.3:6667")); consumerConfig.put(ConsumerConstant.USERNAME_KEY, "root"); consumerConfig.put(ConsumerConstant.PASSWORD_KEY, "root"); try (SubscriptionPullConsumer pullConsumer = new SubscriptionPullConsumer(consumerConfig)) { pullConsumer.open(); pullConsumer.subscribe("my_topic"); while (true) { List messages = pullConsumer.poll(10000); for (final SubscriptionMessage message : messages) { final short messageType = message.getMessageType(); if (SubscriptionMessageType.isValidatedMessageType(messageType)) { for (final SubscriptionSessionDataSet dataSet : message.getSessionDataSetsHandler()) { while (dataSet.hasNext()) { final RowRecord record = dataSet.next(); System.out.println(record); } } } } } } Error: org.apache.iotdb.rpc.subscription.exception.SubscriptionConnectionException: Cluster has no available subscription providers to connect with initial endpoints [TEndPoint(ip:localhost, port:0)] at org.apache.iotdb.session.subscription.consumer.SubscriptionProviders.openProviders(SubscriptionProviders.java:123) at org.apache.iotdb.session.subscription.consumer.SubscriptionConsumer.open(SubscriptionConsumer.java:260) at org.apache.iotdb.session.subscription.consumer.SubscriptionPullConsumer.open(SubscriptionPullConsumer.java:112) at org.apache.iotdb.session.SessionTest.testNullCluster(SessionTest.java:1222) at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:103) at java.base/java.lang.reflect.Method.invoke(Method.java:580) at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:59) at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12) at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:56) at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17) at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26) at org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27) at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306) at org.junit.runners.BlockJUnit4ClassRunner$1.evaluate(BlockJUnit4ClassRunner.java:100) at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:366) at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:103) at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:63) at org.junit.runners.ParentRunner$4.run(ParentRunner.java:331) at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:79) at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:329) at org.junit.runners.ParentRunner.access$100(ParentRunner.java:66) at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:293) at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306) at org.junit.runners.ParentRunner.run(ParentRunner.java:413) at org.junit.runner.JUnitCore.run(JUnitCore.java:137) at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:69) at com.intellij.rt.junit.IdeaTestRunner$Repeater$1.execute(IdeaTestRunner.java:38) at com.intellij.rt.execution.junit.TestsRepeater.repeat(TestsRepeater.java:11) at com.intellij.rt.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:35) at com.intellij.rt.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:232) at com.intellij.rt.junit.JUnitStarter.main(JUnitStarter.java:55) --- .../consumer/base/AbstractSubscriptionConsumer.java | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java index 37c175836622d..56ec3100841f5 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java @@ -189,12 +189,8 @@ protected AbstractSubscriptionConsumer( final AbstractSubscriptionConsumerBuilder builder, final Properties properties) { this( builder - .host( - (String) - properties.getOrDefault(ConsumerConstant.HOST_KEY, SessionConfig.DEFAULT_HOST)) - .port( - (Integer) - properties.getOrDefault(ConsumerConstant.PORT_KEY, SessionConfig.DEFAULT_PORT)) + .host((String) properties.get(ConsumerConstant.HOST_KEY)) + .port((Integer) properties.get(ConsumerConstant.PORT_KEY)) .nodeUrls((List) properties.get(ConsumerConstant.NODE_URLS_KEY)) .username( (String)