Cannot FlinkPulsarSource specify SubscriptionType, or how???

[why198852]

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setParallelism(1)
env.enableCheckpointing(10000)

val prop = new Properties()
prop.put("topic", "persistent://ts/ns/mpc_test")
prop.put("pulsar.reader.subscriptionName", "ts.ns-profile")

val clientConf = new ClientConfigurationData()
clientConf.setAuthPluginClassName("org.apache.pulsar.client.impl.auth.AuthenticationToken")
clientConf.setAuthParams("eDkaliAfeGMsevPF0km6c1B741gxAWT2hgc")
clientConf.setServiceUrl("pulsar://127.0.0.1:6650")

val source1 = new FlinkPulsarSource[String]("http://127.0.0.1:8080", clientConf, new SimpleStringSchema(), prop)
source1.setStartFromSubscription(prop.getProperty("pulsar.reader.subscriptionName", "ts.ns-profile"))
val dd = env.addSource(source1)

dd.print("pulsar:").setParallelism(1)

env.execute(this.getClass.getName)

[jiazhai]

@why198852 currently, we use reader mode instead of consumer to read data. it not support subscriptionType

[sijie]

Close this issue since the question is answered

[ckdarby]

@why198852 currently, we use reader mode instead of consumer to read data. it not support subscriptionType

@jiazhai Is this still the case and is it planned for the future to support consumer with subscription type to read data?