@Test public void testComplexOuputType() throws Exception { PulsarSourceConfig pulsarConfig = getPulsarConfigs(false); pulsarConfig.setTypeClassName(ComplexUserDefinedType.class.getName()); consumerConfigs.put("persistent: ConsumerConfig.builder().serdeClassName(ComplexSerDe.class.getName()).build()); pulsarConfig.setTopicSchema(consumerConfigs); @Cleanup PulsarSource<?> pulsarSource = new PulsarSource<>(getPulsarClient(), pulsarConfig, new HashMap<>(), Thread.currentThread().getContextClassLoader()); pulsarSource.setupConsumerConfigs(); }
@SuppressWarnings("unchecked") @VisibleForTesting Map<String, ConsumerConfig<T>> setupConsumerConfigs() throws ClassNotFoundException { Map<String, ConsumerConfig<T>> configs = new TreeMap<>(); Class<?> typeArg = Reflections.loadClass(this.pulsarSourceConfig.getTypeClassName(), this.functionClassLoader); checkArgument(!Void.class.equals(typeArg), "Input type of Pulsar Function cannot be Void"); pulsarSourceConfig.getTopicSchema().forEach((topic, conf) -> { Schema<T> schema; if (conf.getSerdeClassName() != null && !conf.getSerdeClassName().isEmpty()) { schema = (Schema<T>) topicSchema.getSchema(topic, typeArg, conf.getSerdeClassName(), true); } else { schema = (Schema<T>) topicSchema.getSchema(topic, typeArg, conf, true); } configs.put(topic, ConsumerConfig.<T> builder(). schema(schema). isRegexPattern(conf.isRegexPattern()). receiverQueueSize(conf.getReceiverQueueSize()). consumerProperties(conf.getConsumerProperties()).build()); }); return configs; }
PulsarSource extends PushSource<T> implements MessageListener<T> { @SuppressWarnings("unchecked") @VisibleForTesting Map<String, ConsumerConfig<T>> setupConsumerConfigs() throws ClassNotFoundException { Map<String, ConsumerConfig<T>> configs = new TreeMap<>(); Class<?> typeArg = Reflections.loadClass(this.pulsarSourceConfig.getTypeClassName(), this.functionClassLoader); checkArgument(!Void.class.equals(typeArg), "Input type of Pulsar Function cannot be Void"); pulsarSourceConfig.getTopicSchema().forEach((topic, conf) -> { Schema<T> schema; if (conf.getSerdeClassName() != null && !conf.getSerdeClassName().isEmpty()) { schema = (Schema<T>) topicSchema.getSchema(topic, typeArg, conf.getSerdeClassName(), true); } else { schema = (Schema<T>) topicSchema.getSchema(topic, typeArg, conf, true); } configs.put(topic, ConsumerConfig.<T> builder(). schema(schema). isRegexPattern(conf.isRegexPattern()). receiverQueueSize(conf.getReceiverQueueSize()). consumerProperties(conf.getConsumerProperties()).build()); }); return configs; } }
PulsarSource extends PushSource<T> implements MessageListener<T> { @SuppressWarnings("unchecked") @VisibleForTesting Map<String, ConsumerConfig<T>> setupConsumerConfigs() throws ClassNotFoundException { Map<String, ConsumerConfig<T>> configs = new TreeMap<>(); Class<?> typeArg = Reflections.loadClass(this.pulsarSourceConfig.getTypeClassName(), this.functionClassLoader); checkArgument(!Void.class.equals(typeArg), "Input type of Pulsar Function cannot be Void"); pulsarSourceConfig.getTopicSchema().forEach((topic, conf) -> { Schema<T> schema; if (conf.getSerdeClassName() != null && !conf.getSerdeClassName().isEmpty()) { schema = (Schema<T>) topicSchema.getSchema(topic, typeArg, conf.getSerdeClassName(), true); } else { schema = (Schema<T>) topicSchema.getSchema(topic, typeArg, conf, true); } configs.put(topic, ConsumerConfig.<T> builder(). schema(schema). isRegexPattern(conf.isRegexPattern()). receiverQueueSize(conf.getReceiverQueueSize()). consumerProperties(conf.getConsumerProperties()).build()); }); return configs; } PulsarSource(PulsarClient pulsarClient, PulsarSourceConfig pulsarConfig, Map<String, String> properties,
                        ClassLoader functionClassLoader); }
PulsarSource extends PushSource<T> implements MessageListener<T> { @SuppressWarnings("unchecked") @VisibleForTesting Map<String, ConsumerConfig<T>> setupConsumerConfigs() throws ClassNotFoundException { Map<String, ConsumerConfig<T>> configs = new TreeMap<>(); Class<?> typeArg = Reflections.loadClass(this.pulsarSourceConfig.getTypeClassName(), this.functionClassLoader); checkArgument(!Void.class.equals(typeArg), "Input type of Pulsar Function cannot be Void"); pulsarSourceConfig.getTopicSchema().forEach((topic, conf) -> { Schema<T> schema; if (conf.getSerdeClassName() != null && !conf.getSerdeClassName().isEmpty()) { schema = (Schema<T>) topicSchema.getSchema(topic, typeArg, conf.getSerdeClassName(), true); } else { schema = (Schema<T>) topicSchema.getSchema(topic, typeArg, conf, true); } configs.put(topic, ConsumerConfig.<T> builder(). schema(schema). isRegexPattern(conf.isRegexPattern()). receiverQueueSize(conf.getReceiverQueueSize()). consumerProperties(conf.getConsumerProperties()).build()); }); return configs; } 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> { @SuppressWarnings("unchecked") @VisibleForTesting Map<String, ConsumerConfig<T>> setupConsumerConfigs() throws ClassNotFoundException { Map<String, ConsumerConfig<T>> configs = new TreeMap<>(); Class<?> typeArg = Reflections.loadClass(this.pulsarSourceConfig.getTypeClassName(), this.functionClassLoader); checkArgument(!Void.class.equals(typeArg), "Input type of Pulsar Function cannot be Void"); pulsarSourceConfig.getTopicSchema().forEach((topic, conf) -> { Schema<T> schema; if (conf.getSerdeClassName() != null && !conf.getSerdeClassName().isEmpty()) { schema = (Schema<T>) topicSchema.getSchema(topic, typeArg, conf.getSerdeClassName(), true); } else { schema = (Schema<T>) topicSchema.getSchema(topic, typeArg, conf, true); } configs.put(topic, ConsumerConfig.<T> builder(). schema(schema). isRegexPattern(conf.isRegexPattern()). receiverQueueSize(conf.getReceiverQueueSize()). consumerProperties(conf.getConsumerProperties()).build()); }); return configs; } 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();  }