@Test public void testDefaultSerDe() throws Exception { PulsarSourceConfig pulsarConfig = getPulsarConfigs(false); pulsarConfig.setTypeClassName(String.class.getName()); consumerConfigs.put("persistent: ConsumerConfig.builder().serdeClassName(TopicSchema.DEFAULT_SERDE).build()); pulsarConfig.setTopicSchema(consumerConfigs); @Cleanup PulsarSource<?> pulsarSource = new PulsarSource<>(getPulsarClient(), pulsarConfig, new HashMap<>(), Thread.currentThread().getContextClassLoader()); pulsarSource.open(new HashMap<>(), mock(SourceContext.class)); }
@Override public void open(Map<String, Object> config, SourceContext sourceContext) throws Exception { log.info("Opening pulsar source with config: {}", pulsarSourceConfig); Map<String, ConsumerConfig<T>> configs = setupConsumerConfigs(); for (Map.Entry<String, ConsumerConfig<T>> e : configs.entrySet()) { String topic = e.getKey(); ConsumerConfig<T> conf = e.getValue(); log.info("Creating consumers for topic : {}, schema : {}, schemaInfo: {}", topic, conf.getSchema(), conf.getSchema().getSchemaInfo()); ConsumerBuilder<T> cb = pulsarClient.newConsumer(conf.getSchema()) .cryptoFailureAction(ConsumerCryptoFailureAction.CONSUME) .subscriptionName(pulsarSourceConfig.getSubscriptionName()) .subscriptionInitialPosition(pulsarSourceConfig.getSubscriptionPosition()) .subscriptionType(pulsarSourceConfig.getSubscriptionType()); if (conf.getConsumerProperties() != null && !conf.getConsumerProperties().isEmpty()) { cb.loadConf(new HashMap<>(conf.getConsumerProperties())); } cb.messageListener(this); if (conf.isRegexPattern) { cb = cb.topicsPattern(topic); } else { cb = cb.topics(Collections.singletonList(topic)); } if (conf.getReceiverQueueSize() != null) { cb = cb.receiverQueueSize(conf.getReceiverQueueSize()); } cb = cb.properties(properties); if (pulsarSourceConfig.getNegativeAckRedeliveryDelayMs() != null && pulsarSourceConfig.getNegativeAckRedeliveryDelayMs() > 0) { cb.negativeAckRedeliveryDelay(pulsarSourceConfig.getNegativeAckRedeliveryDelayMs(), TimeUnit.MILLISECONDS); } if (pulsarSourceConfig.getTimeoutMs() != null) { cb = cb.ackTimeout(pulsarSourceConfig.getTimeoutMs(), TimeUnit.MILLISECONDS); } if (pulsarSourceConfig.getMaxMessageRetries() != null && pulsarSourceConfig.getMaxMessageRetries() >= 0) { DeadLetterPolicy.DeadLetterPolicyBuilder deadLetterPolicyBuilder = DeadLetterPolicy.builder(); deadLetterPolicyBuilder.maxRedeliverCount(pulsarSourceConfig.getMaxMessageRetries()); if (pulsarSourceConfig.getDeadLetterTopic() != null && !pulsarSourceConfig.getDeadLetterTopic().isEmpty()) { deadLetterPolicyBuilder.deadLetterTopic(pulsarSourceConfig.getDeadLetterTopic()); } cb = cb.deadLetterPolicy(deadLetterPolicyBuilder.build()); } Consumer<T> consumer = cb.subscribeAsync().join(); inputConsumers.add(consumer); } inputTopics = inputConsumers.stream().flatMap(c -> { return (c instanceof MultiTopicsConsumerImpl) ? ((MultiTopicsConsumerImpl<?>) c).getTopics().stream() : Collections.singletonList(c.getTopic()).stream(); }).collect(Collectors.toList()); }
PulsarSource extends PushSource<T> implements MessageListener<T> { @Override public void open(Map<String, Object> config, SourceContext sourceContext) throws Exception { log.info("Opening pulsar source with config: {}", pulsarSourceConfig); Map<String, ConsumerConfig<T>> configs = setupConsumerConfigs(); for (Map.Entry<String, ConsumerConfig<T>> e : configs.entrySet()) { String topic = e.getKey(); ConsumerConfig<T> conf = e.getValue(); log.info("Creating consumers for topic : {}, schema : {}, schemaInfo: {}", topic, conf.getSchema(), conf.getSchema().getSchemaInfo()); ConsumerBuilder<T> cb = pulsarClient.newConsumer(conf.getSchema()) .cryptoFailureAction(ConsumerCryptoFailureAction.CONSUME) .subscriptionName(pulsarSourceConfig.getSubscriptionName()) .subscriptionInitialPosition(pulsarSourceConfig.getSubscriptionPosition()) .subscriptionType(pulsarSourceConfig.getSubscriptionType()); if (conf.getConsumerProperties() != null && !conf.getConsumerProperties().isEmpty()) { cb.loadConf(new HashMap<>(conf.getConsumerProperties())); } cb.messageListener(this); if (conf.isRegexPattern) { cb = cb.topicsPattern(topic); } else { cb = cb.topics(Collections.singletonList(topic)); } if (conf.getReceiverQueueSize() != null) { cb = cb.receiverQueueSize(conf.getReceiverQueueSize()); } cb = cb.properties(properties); if (pulsarSourceConfig.getNegativeAckRedeliveryDelayMs() != null && pulsarSourceConfig.getNegativeAckRedeliveryDelayMs() > 0) { cb.negativeAckRedeliveryDelay(pulsarSourceConfig.getNegativeAckRedeliveryDelayMs(), TimeUnit.MILLISECONDS); } if (pulsarSourceConfig.getTimeoutMs() != null) { cb = cb.ackTimeout(pulsarSourceConfig.getTimeoutMs(), TimeUnit.MILLISECONDS); } if (pulsarSourceConfig.getMaxMessageRetries() != null && pulsarSourceConfig.getMaxMessageRetries() >= 0) { DeadLetterPolicy.DeadLetterPolicyBuilder deadLetterPolicyBuilder = DeadLetterPolicy.builder(); deadLetterPolicyBuilder.maxRedeliverCount(pulsarSourceConfig.getMaxMessageRetries()); if (pulsarSourceConfig.getDeadLetterTopic() != null && !pulsarSourceConfig.getDeadLetterTopic().isEmpty()) { deadLetterPolicyBuilder.deadLetterTopic(pulsarSourceConfig.getDeadLetterTopic()); } cb = cb.deadLetterPolicy(deadLetterPolicyBuilder.build()); } Consumer<T> consumer = cb.subscribeAsync().join(); inputConsumers.add(consumer); } inputTopics = inputConsumers.stream().flatMap(c -> { return (c instanceof MultiTopicsConsumerImpl) ? ((MultiTopicsConsumerImpl<?>) c).getTopics().stream() : Collections.singletonList(c.getTopic()).stream(); }).collect(Collectors.toList()); } }
PulsarSource extends PushSource<T> implements MessageListener<T> { @Override public void open(Map<String, Object> config, SourceContext sourceContext) throws Exception { log.info("Opening pulsar source with config: {}", pulsarSourceConfig); Map<String, ConsumerConfig<T>> configs = setupConsumerConfigs(); for (Map.Entry<String, ConsumerConfig<T>> e : configs.entrySet()) { String topic = e.getKey(); ConsumerConfig<T> conf = e.getValue(); log.info("Creating consumers for topic : {}, schema : {}, schemaInfo: {}", topic, conf.getSchema(), conf.getSchema().getSchemaInfo()); ConsumerBuilder<T> cb = pulsarClient.newConsumer(conf.getSchema()) .cryptoFailureAction(ConsumerCryptoFailureAction.CONSUME) .subscriptionName(pulsarSourceConfig.getSubscriptionName()) .subscriptionInitialPosition(pulsarSourceConfig.getSubscriptionPosition()) .subscriptionType(pulsarSourceConfig.getSubscriptionType()); if (conf.getConsumerProperties() != null && !conf.getConsumerProperties().isEmpty()) { cb.loadConf(new HashMap<>(conf.getConsumerProperties())); } cb.messageListener(this); if (conf.isRegexPattern) { cb = cb.topicsPattern(topic); } else { cb = cb.topics(Collections.singletonList(topic)); } if (conf.getReceiverQueueSize() != null) { cb = cb.receiverQueueSize(conf.getReceiverQueueSize()); } cb = cb.properties(properties); if (pulsarSourceConfig.getNegativeAckRedeliveryDelayMs() != null && pulsarSourceConfig.getNegativeAckRedeliveryDelayMs() > 0) { cb.negativeAckRedeliveryDelay(pulsarSourceConfig.getNegativeAckRedeliveryDelayMs(), TimeUnit.MILLISECONDS); } if (pulsarSourceConfig.getTimeoutMs() != null) { cb = cb.ackTimeout(pulsarSourceConfig.getTimeoutMs(), TimeUnit.MILLISECONDS); } if (pulsarSourceConfig.getMaxMessageRetries() != null && pulsarSourceConfig.getMaxMessageRetries() >= 0) { DeadLetterPolicy.DeadLetterPolicyBuilder deadLetterPolicyBuilder = DeadLetterPolicy.builder(); deadLetterPolicyBuilder.maxRedeliverCount(pulsarSourceConfig.getMaxMessageRetries()); if (pulsarSourceConfig.getDeadLetterTopic() != null && !pulsarSourceConfig.getDeadLetterTopic().isEmpty()) { deadLetterPolicyBuilder.deadLetterTopic(pulsarSourceConfig.getDeadLetterTopic()); } cb = cb.deadLetterPolicy(deadLetterPolicyBuilder.build()); } Consumer<T> consumer = cb.subscribeAsync().join(); inputConsumers.add(consumer); } inputTopics = inputConsumers.stream().flatMap(c -> { return (c instanceof MultiTopicsConsumerImpl) ? ((MultiTopicsConsumerImpl<?>) c).getTopics().stream() : Collections.singletonList(c.getTopic()).stream(); }).collect(Collectors.toList()); } PulsarSource(PulsarClient pulsarClient, PulsarSourceConfig pulsarConfig, Map<String, String> properties,
                        ClassLoader functionClassLoader); }
PulsarSource extends PushSource<T> implements MessageListener<T> { @Override public void open(Map<String, Object> config, SourceContext sourceContext) throws Exception { log.info("Opening pulsar source with config: {}", pulsarSourceConfig); Map<String, ConsumerConfig<T>> configs = setupConsumerConfigs(); for (Map.Entry<String, ConsumerConfig<T>> e : configs.entrySet()) { String topic = e.getKey(); ConsumerConfig<T> conf = e.getValue(); log.info("Creating consumers for topic : {}, schema : {}, schemaInfo: {}", topic, conf.getSchema(), conf.getSchema().getSchemaInfo()); ConsumerBuilder<T> cb = pulsarClient.newConsumer(conf.getSchema()) .cryptoFailureAction(ConsumerCryptoFailureAction.CONSUME) .subscriptionName(pulsarSourceConfig.getSubscriptionName()) .subscriptionInitialPosition(pulsarSourceConfig.getSubscriptionPosition()) .subscriptionType(pulsarSourceConfig.getSubscriptionType()); if (conf.getConsumerProperties() != null && !conf.getConsumerProperties().isEmpty()) { cb.loadConf(new HashMap<>(conf.getConsumerProperties())); } cb.messageListener(this); if (conf.isRegexPattern) { cb = cb.topicsPattern(topic); } else { cb = cb.topics(Collections.singletonList(topic)); } if (conf.getReceiverQueueSize() != null) { cb = cb.receiverQueueSize(conf.getReceiverQueueSize()); } cb = cb.properties(properties); if (pulsarSourceConfig.getNegativeAckRedeliveryDelayMs() != null && pulsarSourceConfig.getNegativeAckRedeliveryDelayMs() > 0) { cb.negativeAckRedeliveryDelay(pulsarSourceConfig.getNegativeAckRedeliveryDelayMs(), TimeUnit.MILLISECONDS); } if (pulsarSourceConfig.getTimeoutMs() != null) { cb = cb.ackTimeout(pulsarSourceConfig.getTimeoutMs(), TimeUnit.MILLISECONDS); } if (pulsarSourceConfig.getMaxMessageRetries() != null && pulsarSourceConfig.getMaxMessageRetries() >= 0) { DeadLetterPolicy.DeadLetterPolicyBuilder deadLetterPolicyBuilder = DeadLetterPolicy.builder(); deadLetterPolicyBuilder.maxRedeliverCount(pulsarSourceConfig.getMaxMessageRetries()); if (pulsarSourceConfig.getDeadLetterTopic() != null && !pulsarSourceConfig.getDeadLetterTopic().isEmpty()) { deadLetterPolicyBuilder.deadLetterTopic(pulsarSourceConfig.getDeadLetterTopic()); } cb = cb.deadLetterPolicy(deadLetterPolicyBuilder.build()); } Consumer<T> consumer = cb.subscribeAsync().join(); inputConsumers.add(consumer); } inputTopics = inputConsumers.stream().flatMap(c -> { return (c instanceof MultiTopicsConsumerImpl) ? ((MultiTopicsConsumerImpl<?>) c).getTopics().stream() : Collections.singletonList(c.getTopic()).stream(); }).collect(Collectors.toList()); } PulsarSource(PulsarClient pulsarClient, PulsarSourceConfig pulsarConfig, Map<String, String> properties,
                        ClassLoader functionClassLoader); @Override void open(Map<String, Object> config, SourceContext sourceContext); @Override void received(Consumer<T> consumer, Message<T> message); @Override void close(); List<String> getInputTopics(); List<Consumer<T>> getInputConsumers(); }
PulsarSource extends PushSource<T> implements MessageListener<T> { @Override public void open(Map<String, Object> config, SourceContext sourceContext) throws Exception { log.info("Opening pulsar source with config: {}", pulsarSourceConfig); Map<String, ConsumerConfig<T>> configs = setupConsumerConfigs(); for (Map.Entry<String, ConsumerConfig<T>> e : configs.entrySet()) { String topic = e.getKey(); ConsumerConfig<T> conf = e.getValue(); log.info("Creating consumers for topic : {}, schema : {}, schemaInfo: {}", topic, conf.getSchema(), conf.getSchema().getSchemaInfo()); ConsumerBuilder<T> cb = pulsarClient.newConsumer(conf.getSchema()) .cryptoFailureAction(ConsumerCryptoFailureAction.CONSUME) .subscriptionName(pulsarSourceConfig.getSubscriptionName()) .subscriptionInitialPosition(pulsarSourceConfig.getSubscriptionPosition()) .subscriptionType(pulsarSourceConfig.getSubscriptionType()); if (conf.getConsumerProperties() != null && !conf.getConsumerProperties().isEmpty()) { cb.loadConf(new HashMap<>(conf.getConsumerProperties())); } cb.messageListener(this); if (conf.isRegexPattern) { cb = cb.topicsPattern(topic); } else { cb = cb.topics(Collections.singletonList(topic)); } if (conf.getReceiverQueueSize() != null) { cb = cb.receiverQueueSize(conf.getReceiverQueueSize()); } cb = cb.properties(properties); if (pulsarSourceConfig.getNegativeAckRedeliveryDelayMs() != null && pulsarSourceConfig.getNegativeAckRedeliveryDelayMs() > 0) { cb.negativeAckRedeliveryDelay(pulsarSourceConfig.getNegativeAckRedeliveryDelayMs(), TimeUnit.MILLISECONDS); } if (pulsarSourceConfig.getTimeoutMs() != null) { cb = cb.ackTimeout(pulsarSourceConfig.getTimeoutMs(), TimeUnit.MILLISECONDS); } if (pulsarSourceConfig.getMaxMessageRetries() != null && pulsarSourceConfig.getMaxMessageRetries() >= 0) { DeadLetterPolicy.DeadLetterPolicyBuilder deadLetterPolicyBuilder = DeadLetterPolicy.builder(); deadLetterPolicyBuilder.maxRedeliverCount(pulsarSourceConfig.getMaxMessageRetries()); if (pulsarSourceConfig.getDeadLetterTopic() != null && !pulsarSourceConfig.getDeadLetterTopic().isEmpty()) { deadLetterPolicyBuilder.deadLetterTopic(pulsarSourceConfig.getDeadLetterTopic()); } cb = cb.deadLetterPolicy(deadLetterPolicyBuilder.build()); } Consumer<T> consumer = cb.subscribeAsync().join(); inputConsumers.add(consumer); } inputTopics = inputConsumers.stream().flatMap(c -> { return (c instanceof MultiTopicsConsumerImpl) ? ((MultiTopicsConsumerImpl<?>) c).getTopics().stream() : Collections.singletonList(c.getTopic()).stream(); }).collect(Collectors.toList()); } PulsarSource(PulsarClient pulsarClient, PulsarSourceConfig pulsarConfig, Map<String, String> properties,
                        ClassLoader functionClassLoader); @Override void open(Map<String, Object> config, SourceContext sourceContext); @Override void received(Consumer<T> consumer, Message<T> message); @Override void close(); List<String> getInputTopics(); List<Consumer<T>> getInputConsumers();  }