You are viewing a plain text version of this content. The canonical link for it is here.

Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2022/06/13 02:55:39 UTC

[GitHub] [pulsar] xyhmnb opened a new issue, #16029: Help!We have some questions about flink-connector-pulsar.

xyhmnb opened a new issue, #16029:
URL: https://github.com/apache/pulsar/issues/16029

   **Describe the bug**
   these problems about PulsarSource.
   1. use the Exclusive subscribeType .when i start the job ,it can receive messages but not ack messages. i query the topic subscription backlog num is not 0.
   2. use the Shared subscribeType . the flink receive the old messages when i restart the job for each time.it show the topic subscription backlog num is 0,but it still receive the old message when the job restart.
   3. the subscription backlog num is not 0 when i use the new subscriptionName to connect the pulsar. 
   
   
   **To Reproduce**
   1.create service. i use the docker.
   docker run -it -p 6650:6650  -p 8888:8080 --mount source=pulsardata,target=/pulsar/data --mount source=pulsarconf,target=/pulsar/conf apachepulsar/pulsar:2.9.2 bin/pulsar standalone
   2.pulsar2.9.2，flink1.14，flink-connector-pulsar1.14.4
   3.test
   `StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
           PulsarSource<String> source = PulsarSource.builder()
                   .setServiceUrl("pulsar://47.102.192.43:6650")
                   .setAdminUrl("http://47.102.192.43:8888")
                   .setStartCursor(StartCursor.earliest())
                   .setTopics("dev-message")
                   .setDeserializationSchema(PulsarDeserializationSchema.flinkSchema(new SimpleStringSchema()))
                   .setSubscriptionName("flink-subscription")
                   .setSubscriptionType(SubscriptionType.Exclusive)
                   .build();
           DataStreamSource<String> dataStreamSource = env.fromSource(source,
                   WatermarkStrategy.noWatermarks(), "Pulsar Source Work");
           dataStreamSource.print();
           try {
               env.execute("pulsar");
           } catch (Exception e) {
               e.printStackTrace();
           }`
   
   **Expected behavior**
   params:
   serviceUrl: pulsar://47.102.192.43:6650
   adminUrl: http://47.102.192.43:8888
   topic: dev-message
   transactionCoordinatorEnabled: false
   acknowledgmentAtBatchIndexLevelEnabled: false
   
   
   **Screenshots**
   ![image](https://user-images.githubusercontent.com/30762020/173271120-a0c08588-d288-4168-80be-4f52634daaf7.png)
   ![image](https://user-images.githubusercontent.com/30762020/173271177-753374c7-c88d-4e06-ab2f-9d13218d61ee.png)
   the message was print but not ack.
   
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [pulsar] Technoboy- commented on issue #16029: Help!We have some questions about flink-connector-pulsar.

Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.

Technoboy- commented on issue #16029:
URL: https://github.com/apache/pulsar/issues/16029#issuecomment-1430662013

   Fixed by https://github.com/apache/pulsar/pull/16072


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org