diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/io/BatchSourceConfig.java b/pulsar-common/src/main/java/org/apache/pulsar/common/io/BatchSourceConfig.java new file mode 100644 index 0000000000000..9664cecada7ad --- /dev/null +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/io/BatchSourceConfig.java @@ -0,0 +1,50 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.common.io; + +import java.util.Map; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.EqualsAndHashCode; +import lombok.Getter; +import lombok.NoArgsConstructor; +import lombok.Setter; +import lombok.ToString; + +/** + * Pulsar Batch Source configuration. + */ +@Getter +@Setter +@Data +@EqualsAndHashCode +@ToString +@Builder(toBuilder = true) +@NoArgsConstructor +@AllArgsConstructor +public class BatchSourceConfig { + public static final String BATCHSOURCE_CONFIG_KEY = "__BATCHSOURCECONFIGS__"; + public static final String BATCHSOURCE_CLASSNAME_KEY = "__BATCHSOURCECLASSNAME__"; + + // The class used for triggering the discovery process + private String discoveryTriggererClassName; + // The config needed for the discovery Triggerer init + private Map discoveryTriggererConfig; +} diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/io/SourceConfig.java b/pulsar-common/src/main/java/org/apache/pulsar/common/io/SourceConfig.java index 60421bb291aa3..b3a5634304962 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/io/SourceConfig.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/io/SourceConfig.java @@ -63,4 +63,7 @@ public class SourceConfig { // to change behavior at runtime. Currently, this primarily used by the KubernetesManifestCustomizer // interface private String customRuntimeOptions; + + // If this is a BatchSource, its batch related configs are stored here + private BatchSourceConfig batchSourceConfig; } diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java index 38939a6aee4f4..d466bdc27258d 100644 --- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java +++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java @@ -29,8 +29,10 @@ import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.apache.pulsar.common.functions.Resources; +import org.apache.pulsar.common.io.BatchSourceConfig; import org.apache.pulsar.common.io.ConnectorDefinition; import org.apache.pulsar.common.io.SourceConfig; +import org.apache.pulsar.common.naming.TopicDomain; import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.nar.NarClassLoader; import org.apache.pulsar.common.util.ObjectMapperFactory; @@ -89,6 +91,7 @@ public static FunctionDetails convert(SourceConfig sourceConfig, ExtractedSource functionDetailsBuilder.setProcessingGuarantees( convertProcessingGuarantee(sourceConfig.getProcessingGuarantees())); } + // set source spec Function.SourceSpec.Builder sourceSpecBuilder = Function.SourceSpec.newBuilder(); if (sourceDetails.getSourceClassName() != null) { @@ -100,10 +103,21 @@ public static FunctionDetails convert(SourceConfig sourceConfig, ExtractedSource sourceSpecBuilder.setBuiltin(builtin); } + Map configs = new HashMap<>(); if (sourceConfig.getConfigs() != null) { - sourceSpecBuilder.setConfigs(new Gson().toJson(sourceConfig.getConfigs())); + configs.putAll(sourceConfig.getConfigs()); + } + + // Batch source handling + if (sourceConfig.getBatchSourceConfig() != null) { + configs.put(BatchSourceConfig.BATCHSOURCE_CONFIG_KEY, new Gson().toJson(sourceConfig.getBatchSourceConfig())); + configs.put(BatchSourceConfig.BATCHSOURCE_CLASSNAME_KEY, sourceSpecBuilder.getClassName()); + sourceSpecBuilder.setClassName("org.apache.pulsar.io.batch.BatchSourceExecutor"); } + sourceSpecBuilder.setConfigs(new Gson().toJson(configs)); + + if (sourceConfig.getSecrets() != null && !sourceConfig.getSecrets().isEmpty()) { functionDetailsBuilder.setSecretsMap(new Gson().toJson(sourceConfig.getSecrets())); } @@ -167,16 +181,22 @@ public static SourceConfig convertFromDetails(FunctionDetails functionDetails) { if (!StringUtils.isEmpty(sourceSpec.getBuiltin())) { sourceConfig.setArchive("builtin://" + sourceSpec.getBuiltin()); } - if (!StringUtils.isEmpty(sourceSpec.getConfigs())) { - TypeReference> typeRef - = new TypeReference>() {}; - Map configMap; - try { - configMap = ObjectMapperFactory.getThreadLocal().readValue(sourceSpec.getConfigs(), typeRef); - } catch (IOException e) { - log.error("Failed to read configs for source {}", FunctionCommon.getFullyQualifiedName(functionDetails), e); - throw new RuntimeException(e); + Map configMap = extractSourceConfig(sourceSpec, FunctionCommon.getFullyQualifiedName(functionDetails)); + if (configMap != null) { + BatchSourceConfig batchSourceConfig = extractBatchSourceConfig(configMap); + if (batchSourceConfig != null) { + sourceConfig.setBatchSourceConfig(batchSourceConfig); + if (configMap.containsKey(BatchSourceConfig.BATCHSOURCE_CLASSNAME_KEY)) { + if (!StringUtils.isEmpty((String)configMap.get(BatchSourceConfig.BATCHSOURCE_CLASSNAME_KEY))) { + sourceConfig.setClassName((String)configMap.get(BatchSourceConfig.BATCHSOURCE_CLASSNAME_KEY)); + } else { + sourceConfig.setClassName(null); + } + } } + + configMap.remove(BatchSourceConfig.BATCHSOURCE_CONFIG_KEY); + configMap.remove(BatchSourceConfig.BATCHSOURCE_CLASSNAME_KEY); sourceConfig.setConfigs(configMap); } if (!isEmpty(functionDetails.getSecretsMap())) { @@ -346,6 +366,10 @@ public static ExtractedSourceDetails validate(SourceConfig sourceConfig, Path ar ValidatorUtils.validateSchema(sourceConfig.getSchemaType(), typeArg, classLoader, false); } + if (sourceConfig.getBatchSourceConfig() != null) { + validateBatchSourceConfig(sourceConfig.getBatchSourceConfig()); + } + return new ExtractedSourceDetails(sourceClassName, typeArg.getName()); } @@ -396,9 +420,86 @@ public static SourceConfig validateUpdate(SourceConfig existingConfig, SourceCon if (!StringUtils.isEmpty(newConfig.getCustomRuntimeOptions())) { mergedConfig.setCustomRuntimeOptions(newConfig.getCustomRuntimeOptions()); } + if (isBatchSource(existingConfig) != isBatchSource(newConfig)) { + throw new IllegalArgumentException("Sources cannot be update between regular sources and batchsource"); + } + if (newConfig.getBatchSourceConfig() != null) { + validateBatchSourceConfigUpdate(existingConfig.getBatchSourceConfig(), newConfig.getBatchSourceConfig()); + mergedConfig.setBatchSourceConfig(newConfig.getBatchSourceConfig()); + } return mergedConfig; } + public static void validateBatchSourceConfig(BatchSourceConfig batchSourceConfig) throws IllegalArgumentException { + if (isEmpty(batchSourceConfig.getDiscoveryTriggererClassName())) { + log.error("BatchSourceConfig does not specify Discovery Trigger ClassName"); + throw new IllegalArgumentException("BatchSourceConfig does not specify Discovery Trigger ClassName"); + } + } + + public static Map extractSourceConfig(Function.SourceSpec sourceSpec, String fqfn) { + if (!StringUtils.isEmpty(sourceSpec.getConfigs())) { + TypeReference> typeRef + = new TypeReference>() { + }; + try { + return ObjectMapperFactory.getThreadLocal().readValue(sourceSpec.getConfigs(), typeRef); + } catch (IOException e) { + log.error("Failed to read configs for source {}", fqfn, e); + throw new RuntimeException(e); + } + } else { + return null; + } + } + + public static BatchSourceConfig extractBatchSourceConfig(Map configMap) { + if (configMap.containsKey(BatchSourceConfig.BATCHSOURCE_CONFIG_KEY)) { + String batchSourceConfigJson = (String) configMap.get(BatchSourceConfig.BATCHSOURCE_CONFIG_KEY); + return new Gson().fromJson(batchSourceConfigJson, BatchSourceConfig.class); + } else { + return null; + } + } + + public static Map computeBatchSourceIntermediateTopicSubscriptions(Function.FunctionDetails details, + String fqfn) { + Map configMap = extractSourceConfig(details.getSource(), fqfn); + if (configMap != null) { + BatchSourceConfig batchSourceConfig = extractBatchSourceConfig(configMap); + String intermediateTopicName = computeBatchSourceIntermediateTopicName(details.getTenant(), + details.getNamespace(), details.getName()).toString(); + if (batchSourceConfig != null) { + Map subscriptionMap = new HashMap<>(); + subscriptionMap.put(intermediateTopicName, + computeBatchSourceInstanceSubscriptionName(details.getTenant(), + details.getNamespace(), details.getName())); + return subscriptionMap; + } + } + return null; + } + + public static String computeBatchSourceInstanceSubscriptionName(String tenant, String namespace, + String sourceName) { + return "BatchSourceExecutor-" + tenant + "/" + namespace + "/" + sourceName; + } + + public static TopicName computeBatchSourceIntermediateTopicName(String tenant, String namespace, + String sourceName) { + return TopicName.get(TopicDomain.persistent.name(), tenant, namespace, sourceName + "-intermediate"); + } + + public static boolean isBatchSource(SourceConfig sourceConfig) { + return sourceConfig.getBatchSourceConfig() != null; + } + + public static void validateBatchSourceConfigUpdate(BatchSourceConfig existingConfig, BatchSourceConfig newConfig) { + if (!existingConfig.getDiscoveryTriggererClassName().equals(newConfig.getDiscoveryTriggererClassName())) { + throw new IllegalArgumentException("DiscoverTriggerer class cannot be updated for batchsources"); + } + } + public static void validateConnectorConfig(SourceConfig sourceConfig, ClassLoader classLoader) { try { ConnectorDefinition defn = ConnectorUtils.getConnectorDefinition(classLoader); @@ -417,5 +518,4 @@ public static void validateConnectorConfig(SourceConfig sourceConfig, ClassLoade throw new IllegalArgumentException("Could not validate source config: " + e.getMessage()); } } - } diff --git a/pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/SourceConfigUtilsTest.java b/pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/SourceConfigUtilsTest.java index 4c80b236281bd..ad40b8392619f 100644 --- a/pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/SourceConfigUtilsTest.java +++ b/pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/SourceConfigUtilsTest.java @@ -24,6 +24,7 @@ import lombok.experimental.Accessors; import org.apache.pulsar.common.functions.FunctionConfig; import org.apache.pulsar.common.functions.Resources; +import org.apache.pulsar.common.io.BatchSourceConfig; import org.apache.pulsar.common.io.ConnectorDefinition; import org.apache.pulsar.common.io.SinkConfig; import org.apache.pulsar.common.io.SourceConfig; @@ -31,6 +32,8 @@ import org.apache.pulsar.functions.api.utils.IdentityFunction; import org.apache.pulsar.functions.proto.Function; import org.apache.pulsar.functions.utils.io.ConnectorUtils; +import org.apache.pulsar.io.core.BatchSourceTriggerer; +import org.apache.pulsar.io.core.SourceContext; import org.powermock.api.mockito.PowerMockito; import org.powermock.core.classloader.annotations.PowerMockIgnore; import org.powermock.core.classloader.annotations.PrepareForTest; @@ -41,6 +44,7 @@ import java.lang.reflect.Field; import java.util.HashMap; import java.util.Map; +import java.util.function.Consumer; import static org.apache.pulsar.common.functions.FunctionConfig.ProcessingGuarantees.EFFECTIVELY_ONCE; import static org.mockito.ArgumentMatchers.any; @@ -64,27 +68,41 @@ public static class TestSourceConfig { private String configParameter; } + class TestTriggerer implements BatchSourceTriggerer { + + @Override + public void init(Map config, SourceContext sourceContext) throws Exception { + + } + + @Override + public void start(Consumer trigger) { + + } + + @Override + public void stop() { + + } + } + @Test public void testConvertBackFidelity() throws IOException { - SourceConfig sourceConfig = new SourceConfig(); - sourceConfig.setTenant("test-tenant"); - sourceConfig.setNamespace("test-namespace"); - sourceConfig.setName("test-source"); - sourceConfig.setArchive("builtin://jdbc"); - sourceConfig.setTopicName("test-output"); - sourceConfig.setSerdeClassName("test-serde"); - sourceConfig.setParallelism(1); - sourceConfig.setRuntimeFlags("-DKerberos"); - sourceConfig.setProcessingGuarantees(FunctionConfig.ProcessingGuarantees.ATLEAST_ONCE); + SourceConfig sourceConfig = createSourceConfig(); + Function.FunctionDetails functionDetails = SourceConfigUtils.convert(sourceConfig, new SourceConfigUtils.ExtractedSourceDetails(null, null)); + SourceConfig convertedConfig = SourceConfigUtils.convertFromDetails(functionDetails); - Map consumerConfigs = new HashMap<>(); - consumerConfigs.put("security.protocal", "SASL_PLAINTEXT"); - Map configs = new HashMap<>(); - configs.put("topic", "kafka"); - configs.put("bootstrapServers", "server-1,server-2"); - configs.put("consumerConfigProperties", consumerConfigs); + // add default resources + sourceConfig.setResources(Resources.getDefaultResources()); + assertEquals( + new Gson().toJson(sourceConfig), + new Gson().toJson(convertedConfig) + ); + } - sourceConfig.setConfigs(configs); + @Test + public void testConvertBackFidelityWithBatch() throws IOException { + SourceConfig sourceConfig = createSourceConfigWithBatch(); Function.FunctionDetails functionDetails = SourceConfigUtils.convert(sourceConfig, new SourceConfigUtils.ExtractedSourceDetails(null, null)); SourceConfig convertedConfig = SourceConfigUtils.convertFromDetails(functionDetails); @@ -107,6 +125,17 @@ public void testMergeEqual() { ); } + @Test + public void testBatchConfigMergeEqual() { + SourceConfig sourceConfig = createSourceConfigWithBatch(); + SourceConfig newSourceConfig = createSourceConfigWithBatch(); + SourceConfig mergedConfig = SourceConfigUtils.validateUpdate(sourceConfig, newSourceConfig); + assertEquals( + new Gson().toJson(sourceConfig), + new Gson().toJson(mergedConfig) + ); + } + @Test(expectedExceptions = IllegalArgumentException.class, expectedExceptionsMessageRegExp = "Function Names differ") public void testMergeDifferentName() { SourceConfig sourceConfig = createSourceConfig(); @@ -148,7 +177,7 @@ public void testMergeDifferentClassName() { public void testMergeDifferentProcessingGuarantees() { SourceConfig sourceConfig = createSourceConfig(); SourceConfig newSourceConfig = createUpdatedSourceConfig("processingGuarantees", EFFECTIVELY_ONCE); - SourceConfigUtils.validateUpdate(sourceConfig, newSourceConfig); + SourceConfig mergedConfig = SourceConfigUtils.validateUpdate(sourceConfig, newSourceConfig); } @Test @@ -238,6 +267,35 @@ public void testMergeRuntimeFlags() { ); } + @Test(expectedExceptions = IllegalArgumentException.class, expectedExceptionsMessageRegExp = "DiscoverTriggerer class cannot be updated for batchsources") + public void testMergeDifferentBatchTriggerer() { + SourceConfig sourceConfig = createSourceConfigWithBatch(); + BatchSourceConfig batchSourceConfig = createBatchSourceConfig(); + batchSourceConfig.setDiscoveryTriggererClassName("SomeOtherClassName"); + SourceConfig newSourceConfig = createUpdatedSourceConfig("batchSourceConfig", batchSourceConfig); + SourceConfigUtils.validateUpdate(sourceConfig, newSourceConfig); + } + + @Test + public void testMergeDifferentBatchSourceConfig() { + SourceConfig sourceConfig = createSourceConfigWithBatch(); + BatchSourceConfig batchSourceConfig = createBatchSourceConfig(); + Map newConfig = new HashMap<>(); + newConfig.put("something", "different"); + batchSourceConfig.setDiscoveryTriggererConfig(newConfig); + SourceConfig newSourceConfig = createUpdatedSourceConfig("batchSourceConfig", batchSourceConfig); + SourceConfig mergedConfig = SourceConfigUtils.validateUpdate(sourceConfig, newSourceConfig); + assertEquals( + mergedConfig.getBatchSourceConfig().getDiscoveryTriggererConfig().get("something"), + "different" + ); + mergedConfig.getBatchSourceConfig().setDiscoveryTriggererConfig(sourceConfig.getBatchSourceConfig().getDiscoveryTriggererConfig()); + assertEquals( + new Gson().toJson(sourceConfig), + new Gson().toJson(mergedConfig) + ); + } + @Test public void testValidateConfig() throws IOException { mockStatic(ConnectorUtils.class); @@ -257,17 +315,42 @@ public void testValidateConfig() throws IOException { assertTrue(e.getMessage().contains("Could not validate source config: Field 'configParameter' cannot be null!")); } + private SourceConfig createSourceConfigWithBatch() { + SourceConfig sourceConfig = createSourceConfig(); + BatchSourceConfig batchSourceConfig = createBatchSourceConfig(); + sourceConfig.setBatchSourceConfig(batchSourceConfig); + return sourceConfig; + } + + private BatchSourceConfig createBatchSourceConfig() { + BatchSourceConfig batchSourceConfig = new BatchSourceConfig(); + batchSourceConfig.setDiscoveryTriggererClassName(TestTriggerer.class.getName()); + Map batchConfig = new HashMap<>(); + batchConfig.put("foo", "bar"); + batchSourceConfig.setDiscoveryTriggererConfig(batchConfig); + return batchSourceConfig; + } + private SourceConfig createSourceConfig() { SourceConfig sourceConfig = new SourceConfig(); sourceConfig.setTenant("test-tenant"); sourceConfig.setNamespace("test-namespace"); sourceConfig.setName("test-source"); - sourceConfig.setParallelism(1); - sourceConfig.setClassName(IdentityFunction.class.getName()); + sourceConfig.setArchive("builtin://jdbc"); sourceConfig.setTopicName("test-output"); sourceConfig.setSerdeClassName("test-serde"); + sourceConfig.setParallelism(1); + sourceConfig.setRuntimeFlags("-DKerberos"); sourceConfig.setProcessingGuarantees(FunctionConfig.ProcessingGuarantees.ATLEAST_ONCE); - sourceConfig.setConfigs(new HashMap<>()); + + Map consumerConfigs = new HashMap<>(); + consumerConfigs.put("security.protocal", "SASL_PLAINTEXT"); + Map configs = new HashMap<>(); + configs.put("topic", "kafka"); + configs.put("bootstrapServers", "server-1,server-2"); + configs.put("consumerConfigProperties", consumerConfigs); + + sourceConfig.setConfigs(configs); return sourceConfig; } diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java index 0a9575568a9c8..ad93ba211a3ce 100644 --- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java +++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java @@ -43,6 +43,7 @@ import org.apache.pulsar.functions.runtime.RuntimeSpawner; import org.apache.pulsar.functions.utils.Actions; import org.apache.pulsar.functions.utils.FunctionCommon; +import org.apache.pulsar.functions.utils.SourceConfigUtils; import org.apache.pulsar.functions.utils.io.ConnectorUtils; import java.io.File; @@ -323,63 +324,79 @@ public void accept(Map.Entry stringConsumerSpecEn ? InstanceUtils.getDefaultSubscriptionName(functionRuntimeInfo.getFunctionInstance().getFunctionMetaData().getFunctionDetails()) : functionRuntimeInfo.getFunctionInstance().getFunctionMetaData().getFunctionDetails().getSource().getSubscriptionName(); - try { - Actions.newBuilder() - .addAction( - Actions.Action.builder() - .actionName(String.format("Cleaning up subscriptions for function %s", fqfn)) - .numRetries(10) - .sleepBetweenInvocationsMs(1000) - .supplier(() -> { - try { - if (consumerSpec.getIsRegexPattern()) { - pulsarAdmin.namespaces().unsubscribeNamespace(TopicName - .get(topic).getNamespace(), subscriptionName); - } else { - pulsarAdmin.topics().deleteSubscription(topic, - subscriptionName); - } - } catch (PulsarAdminException e) { - if (e instanceof PulsarAdminException.NotFoundException) { - return Actions.ActionResult.builder() - .success(true) - .build(); - } else { - // for debugging purposes - List> existingConsumers = Collections.emptyList(); - try { - TopicStats stats = pulsarAdmin.topics().getStats(topic); - SubscriptionStats sub = stats.subscriptions.get(subscriptionName); - if (sub != null) { - existingConsumers = sub.consumers.stream() - .map(consumerStats -> consumerStats.metadata) - .collect(Collectors.toList()); - } - } catch (PulsarAdminException e1) { - - } - - String errorMsg = e.getHttpError() != null ? e.getHttpError() : e.getMessage(); - return Actions.ActionResult.builder() - .success(false) - .errorMsg(String.format("%s - existing consumers: %s", errorMsg, existingConsumers)) - .build(); - } - } - - return Actions.ActionResult.builder() - .success(true) - .build(); - - }) - .build()) - .run(); - } catch (InterruptedException e) { - throw new RuntimeException(e); - } + deleteSubscription(topic, consumerSpec, subscriptionName, fqfn); } }); } + if (InstanceUtils.calculateSubjectType(details) == FunctionDetails.ComponentType.SOURCE) { + // topicName -> subscriptions + Map subscriptions = + SourceConfigUtils.computeBatchSourceIntermediateTopicSubscriptions(details, + FunctionCommon.getFullyQualifiedName(details)); + if (subscriptions != null) { + subscriptions.forEach((topic, subscriptionName) -> { + Function.ConsumerSpec consumerSpec = Function.ConsumerSpec.newBuilder().setIsRegexPattern(false).build(); + deleteSubscription(topic, consumerSpec, subscriptionName, fqfn); + }); + } + } + } + + private void deleteSubscription(String topic, Function.ConsumerSpec consumerSpec, String subscriptionName, String fqfn) { + try { + Actions.newBuilder() + .addAction( + Actions.Action.builder() + .actionName(String.format("Cleaning up subscriptions for function %s", fqfn)) + .numRetries(10) + .sleepBetweenInvocationsMs(1000) + .supplier(() -> { + try { + if (consumerSpec.getIsRegexPattern()) { + pulsarAdmin.namespaces().unsubscribeNamespace(TopicName + .get(topic).getNamespace(), subscriptionName); + } else { + pulsarAdmin.topics().deleteSubscription(topic, + subscriptionName); + } + } catch (PulsarAdminException e) { + if (e instanceof PulsarAdminException.NotFoundException) { + return Actions.ActionResult.builder() + .success(true) + .build(); + } else { + // for debugging purposes + List> existingConsumers = Collections.emptyList(); + try { + TopicStats stats = pulsarAdmin.topics().getStats(topic); + SubscriptionStats sub = stats.subscriptions.get(subscriptionName); + if (sub != null) { + existingConsumers = sub.consumers.stream() + .map(consumerStats -> consumerStats.metadata) + .collect(Collectors.toList()); + } + } catch (PulsarAdminException e1) { + + } + + String errorMsg = e.getHttpError() != null ? e.getHttpError() : e.getMessage(); + return Actions.ActionResult.builder() + .success(false) + .errorMsg(String.format("%s - existing consumers: %s", errorMsg, existingConsumers)) + .build(); + } + } + + return Actions.ActionResult.builder() + .success(true) + .build(); + + }) + .build()) + .run(); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } } private String getDownloadPackagePath(FunctionMetaData functionMetaData, int instanceId) { diff --git a/pulsar-io/batch-data-generator/pom.xml b/pulsar-io/batch-data-generator/pom.xml new file mode 100644 index 0000000000000..9573cd75b80e6 --- /dev/null +++ b/pulsar-io/batch-data-generator/pom.xml @@ -0,0 +1,88 @@ + + + 4.0.0 + + org.apache.pulsar + pulsar-io + 2.6.0-SNAPSHOT + + + pulsar-io-batch-data-generator + Pulsar IO :: Batch Data Generator + + + + + ${project.groupId} + pulsar-io-core + ${project.version} + + + + ${project.groupId} + pulsar-common + ${project.version} + + + + ${project.groupId} + pulsar-io-batch-discovery-triggerers + ${project.version} + + + + ${project.groupId} + pulsar-io-batch + ${project.version} + + + + io.codearte.jfairy + jfairy + 0.5.9 + + + + org.apache.avro + avro + ${avro.version} + + + + ${project.groupId} + pulsar-functions-local-runner-original + ${project.version} + test + + + + + + + + org.apache.nifi + nifi-nar-maven-plugin + + + + diff --git a/pulsar-io/batch-data-generator/src/main/java/org/apache/pulsar/io/batchdatagenerator/BatchDataGeneratorPrintSink.java b/pulsar-io/batch-data-generator/src/main/java/org/apache/pulsar/io/batchdatagenerator/BatchDataGeneratorPrintSink.java new file mode 100644 index 0000000000000..2d09c5658dbb4 --- /dev/null +++ b/pulsar-io/batch-data-generator/src/main/java/org/apache/pulsar/io/batchdatagenerator/BatchDataGeneratorPrintSink.java @@ -0,0 +1,46 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.io.batchdatagenerator; + +import lombok.extern.slf4j.Slf4j; +import org.apache.pulsar.functions.api.Record; +import org.apache.pulsar.io.core.Sink; +import org.apache.pulsar.io.core.SinkContext; + +import java.util.Map; + +@Slf4j +public class BatchDataGeneratorPrintSink implements Sink { + + @Override + public void open(Map config, SinkContext sinkContext) { + + } + + @Override + public void write(Record record) { + log.info("RECV: {}", record.getValue()); + record.ack(); + } + + @Override + public void close() { + + } +} diff --git a/pulsar-io/batch-data-generator/src/main/java/org/apache/pulsar/io/batchdatagenerator/BatchDataGeneratorSource.java b/pulsar-io/batch-data-generator/src/main/java/org/apache/pulsar/io/batchdatagenerator/BatchDataGeneratorSource.java new file mode 100644 index 0000000000000..bee005d9e895d --- /dev/null +++ b/pulsar-io/batch-data-generator/src/main/java/org/apache/pulsar/io/batchdatagenerator/BatchDataGeneratorSource.java @@ -0,0 +1,85 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.io.batchdatagenerator; + +import io.codearte.jfairy.Fairy; +import lombok.extern.slf4j.Slf4j; +import org.apache.pulsar.functions.api.Record; +import org.apache.pulsar.io.core.BatchSource; +import org.apache.pulsar.io.core.SourceContext; + +import java.util.Map; +import java.util.Optional; +import java.util.function.Consumer; + +@Slf4j +public class BatchDataGeneratorSource implements BatchSource { + + private Fairy fairy; + private SourceContext sourceContext; + int iteration; + int maxRecordsPerCycle = 10; + + @Override + public void open(Map config, SourceContext sourceContext) { + this.fairy = Fairy.create(); + this.sourceContext = sourceContext; + } + + @Override + public void discover(Consumer taskEater) { + log.info("Generating one task for each instance"); + for (int i = 0; i < sourceContext.getNumInstances(); ++i) { + taskEater.accept("something".getBytes()); + } + } + + @Override + public void prepare(byte[] instanceSplit) { + log.info("Instance " + sourceContext.getInstanceId() + " got a new discovered task"); + final String str = new String(instanceSplit); + final String expected = "something"; + assert str.equals(expected); + iteration = 0; + } + + @Override + public Record readNext() throws Exception { + if (iteration++ < maxRecordsPerCycle) { + Thread.sleep(50); + return new Record() { + @Override + public Optional getKey() { + return Optional.empty(); + } + + @Override + public Person getValue() { + return new Person(fairy.person()); + } + }; + } + return null; + } + + @Override + public void close() { + + } +} diff --git a/pulsar-io/batch-data-generator/src/main/java/org/apache/pulsar/io/batchdatagenerator/Person.java b/pulsar-io/batch-data-generator/src/main/java/org/apache/pulsar/io/batchdatagenerator/Person.java new file mode 100644 index 0000000000000..0015dabea7b2b --- /dev/null +++ b/pulsar-io/batch-data-generator/src/main/java/org/apache/pulsar/io/batchdatagenerator/Person.java @@ -0,0 +1,113 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.io.batchdatagenerator; + +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.NoArgsConstructor; + +@Data +@AllArgsConstructor +@NoArgsConstructor +/** + * This class serves as a copy of of io.codearte.jfairy.producer.person.Person + * because io.codearte.jfairy.producer.person.Person does not + * have default constructors needed to deserialize POJOs + */ +public class Person { + private Address address; + private String firstName; + private String middleName; + private String lastName; + private String email; + private String username; + private String password; + private Sex sex; + private String telephoneNumber; + @org.apache.avro.reflect.AvroSchema("{ \"type\": \"long\", \"logicalType\": \"timestamp-millis\" }") + private long dateOfBirth; + private Integer age; + private Company company; + private String companyEmail; + private String nationalIdentityCardNumber; + private String nationalIdentificationNumber; + private String passportNumber; + + public enum Sex { + MALE, + FEMALE; + + private Sex() { + } + } + + public Person(io.codearte.jfairy.producer.person.Person person) { + this(new Address(person.getAddress()), + person.getFirstName(), + person.getMiddleName(), + person.getLastName(), + person.getEmail(), + person.getUsername(), + person.getPassword(), + Sex.valueOf(person.getSex().name()), + person.getTelephoneNumber(), + person.getDateOfBirth().getMillis(), + person.getAge(), + new Company(person.getCompany()), + person.getCompanyEmail(), + person.getNationalIdentityCardNumber(), + person.getNationalIdentificationNumber(), + person.getPassportNumber()); + } + + @Data + @AllArgsConstructor + @NoArgsConstructor + public static class Company { + private String name; + private String domain; + private String email; + private String vatIdentificationNumber; + public Company(io.codearte.jfairy.producer.company.Company company) { + this(company.getName(), + company.getDomain(), + company.getEmail(), + company.getVatIdentificationNumber()); + } + } + + @Data + @AllArgsConstructor + @NoArgsConstructor + public static class Address { + protected String street; + protected String streetNumber; + protected String apartmentNumber; + protected String postalCode; + protected String city; + + public Address(io.codearte.jfairy.producer.person.Address address) { + this(address.getStreet(), + address.getStreetNumber(), + address.getApartmentNumber(), + address.getPostalCode(), + address.getCity()); + } + } +} diff --git a/pulsar-io/batch-data-generator/src/main/resources/META-INF/services/pulsar-io.yaml b/pulsar-io/batch-data-generator/src/main/resources/META-INF/services/pulsar-io.yaml new file mode 100644 index 0000000000000..03696c904de52 --- /dev/null +++ b/pulsar-io/batch-data-generator/src/main/resources/META-INF/services/pulsar-io.yaml @@ -0,0 +1,23 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# + +name: batch-data-generator +description: Test batch data generator source +sourceClass: org.apache.pulsar.io.batchdatagenerator.BatchDataGeneratorSource +sinkClass: org.apache.pulsar.io.batchdatagenerator.BatchDataGeneratorPrintSink diff --git a/pulsar-io/batch-data-generator/src/test/java/org/apache/pulsar/io/batchdatagenerator/BatchDataGeneratorExec.java b/pulsar-io/batch-data-generator/src/test/java/org/apache/pulsar/io/batchdatagenerator/BatchDataGeneratorExec.java new file mode 100644 index 0000000000000..1e7ed7ffd3e94 --- /dev/null +++ b/pulsar-io/batch-data-generator/src/test/java/org/apache/pulsar/io/batchdatagenerator/BatchDataGeneratorExec.java @@ -0,0 +1,70 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.io.batchdatagenerator; + +import org.apache.pulsar.common.io.BatchSourceConfig; +import org.apache.pulsar.common.io.SourceConfig; +import org.apache.pulsar.functions.LocalRunner; +import org.apache.pulsar.io.batchdiscovery.CronTriggerer; + +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +/** + * Useful for testing within IDE. + * + */ +public class BatchDataGeneratorExec { + + public static void main(final String[] args) throws Exception { + + final String cronString = "0 0/5 * * * ?"; + final Map discoveryConfig = new HashMap<>(); + discoveryConfig.put(CronTriggerer.CRON_KEY, cronString); + + final BatchSourceConfig batchSourceConfig = + BatchSourceConfig.builder() + .discoveryTriggererClassName(CronTriggerer.class.getName()) + .discoveryTriggererConfig(discoveryConfig) + .build(); + + final SourceConfig sourceConfig = + SourceConfig.builder() + .batchSourceConfig(batchSourceConfig) + .className(BatchDataGeneratorSource.class.getName()) + .configs(new HashMap<>()) + .name("BatchDataGenerator") + .parallelism(1) + .topicName("persistent://public/default/batchdatagenerator") + .build(); + + final LocalRunner localRunner = + LocalRunner.builder() + .brokerServiceUrl("pulsar://localhost:6650") + .sourceConfig(sourceConfig) + .build(); + + localRunner.start(false); + TimeUnit.MINUTES.sleep(30); + localRunner.stop(); + + System.exit(0); + } + } diff --git a/pulsar-io/batch-discovery-triggerers/pom.xml b/pulsar-io/batch-discovery-triggerers/pom.xml new file mode 100644 index 0000000000000..b369e18954886 --- /dev/null +++ b/pulsar-io/batch-discovery-triggerers/pom.xml @@ -0,0 +1,55 @@ + + + 4.0.0 + + + org.apache.pulsar + pulsar-io + 2.6.0-SNAPSHOT + + + pulsar-io-batch-discovery-triggerers + Pulsar IO :: Batch Discovery Triggerers + + + + ${project.groupId} + pulsar-io-core + ${project.version} + + + + com.cronutils + cron-utils + 9.0.1 + + + + org.springframework + spring-context + 5.2.5.RELEASE + + + + + diff --git a/pulsar-io/batch-discovery-triggerers/src/main/java/org/apache/pulsar/io/batchdiscovery/CronTriggerer.java b/pulsar-io/batch-discovery-triggerers/src/main/java/org/apache/pulsar/io/batchdiscovery/CronTriggerer.java new file mode 100644 index 0000000000000..6c35b9d68448c --- /dev/null +++ b/pulsar-io/batch-discovery-triggerers/src/main/java/org/apache/pulsar/io/batchdiscovery/CronTriggerer.java @@ -0,0 +1,65 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.io.batchdiscovery; + +import lombok.extern.slf4j.Slf4j; +import org.apache.pulsar.io.core.*; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import org.springframework.scheduling.support.CronTrigger; + +import java.util.Map; +import java.util.Objects; +import java.util.function.Consumer; + +/** + * This is an implementation of BatchSourceTriggerer that triggers based on a cron expression. + * BatchSource developers using this should pass the json string of a map that contains + * "__CRON__" key with the appropriate cron expression. The triggerer will trigger based on this expression. + * + */ +@Slf4j +public class CronTriggerer implements BatchSourceTriggerer { + public static final String CRON_KEY = "__CRON__"; + private String cronExpression; + private ThreadPoolTaskScheduler scheduler; + + @Override + public void init(Map config, SourceContext sourceContext) { + if (config == null || config.containsKey(CRON_KEY)) { + cronExpression = (String) Objects.requireNonNull(config).get(CRON_KEY); + } else { + throw new IllegalArgumentException("Cron Trigger is not provided with Cron String"); + } + log.info("Initialized CronTrigger with expression: {}", cronExpression); + } + + @Override + public void start(Consumer trigger) { + scheduler = new ThreadPoolTaskScheduler(); + scheduler.initialize(); + scheduler.schedule(() -> trigger.accept("CRON"), new CronTrigger(cronExpression)); + } + + @Override + public void stop() { + scheduler.shutdown(); + } + +} + diff --git a/pulsar-io/batch/pom.xml b/pulsar-io/batch/pom.xml new file mode 100644 index 0000000000000..84d22e0e52ec1 --- /dev/null +++ b/pulsar-io/batch/pom.xml @@ -0,0 +1,52 @@ + + + 4.0.0 + + org.apache.pulsar + pulsar-io + 2.6.0-SNAPSHOT + + + pulsar-io-batch + Pulsar IO :: Batch + + + + ${project.groupId} + pulsar-io-core + ${project.version} + + + ${project.groupId} + pulsar-common + ${project.version} + + + ${project.groupId} + pulsar-functions-utils + ${project.version} + + + + + diff --git a/pulsar-io/batch/src/main/java/org/apache/pulsar/io/batch/BatchSourceExecutor.java b/pulsar-io/batch/src/main/java/org/apache/pulsar/io/batch/BatchSourceExecutor.java new file mode 100644 index 0000000000000..88a011e324bc6 --- /dev/null +++ b/pulsar-io/batch/src/main/java/org/apache/pulsar/io/batch/BatchSourceExecutor.java @@ -0,0 +1,240 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.io.batch; + +import com.google.gson.Gson; +import lombok.extern.slf4j.Slf4j; +import org.apache.pulsar.client.api.*; +import org.apache.pulsar.common.io.BatchSourceConfig; +import org.apache.pulsar.functions.api.Record; +import org.apache.pulsar.functions.utils.Actions; +import org.apache.pulsar.functions.utils.FunctionCommon; +import org.apache.pulsar.functions.utils.Reflections; +import org.apache.pulsar.functions.utils.SourceConfigUtils; +import org.apache.pulsar.io.core.*; + +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.CompletableFuture; + +/** + * BatchSourceExecutor wraps BatchSource as Source. Thus from Pulsar IO perspective, it is running a regular + * streaming source. The BatchSourceExecutor orchestrates the lifecycle of BatchSource. + * + * The current implementation uses an intermediate topic between the discovery process and the actual batchsource + * instances. The Discovery is run on 0th instance. Any tasks discovered during the discover are written to the + * intermediate topic. All the instances consume tasks from this intermediate topic using a shared subscription. + */ + +@Slf4j +public class BatchSourceExecutor implements Source { + + private Map config; + private SourceContext sourceContext; + private BatchSourceTriggerer discoveryTriggerer; // Only init in instance 0 + private Consumer intermediateTopicConsumer; + private Message currentTask; + private BatchSourceConfig batchSourceConfig; + private String batchSourceClassName; + private BatchSource batchSource; + private String intermediateTopicName; + + @Override + public void open(Map config, SourceContext sourceContext) throws Exception { + this.config = config; + this.sourceContext = sourceContext; + this.intermediateTopicName = SourceConfigUtils.computeBatchSourceIntermediateTopicName(sourceContext.getTenant(), + sourceContext.getNamespace(), sourceContext.getSourceName()).toString(); + this.getBatchSourceConfigs(config); + this.initializeBatchSource(); + this.start(); + } + + @Override + public Record read() throws Exception { + while (true) { + if (currentTask == null) { + retrieveNextTask(); + prepareInternal(); + } + Record retval = batchSource.readNext(); + if (retval == null) { + // signals end if this batch + intermediateTopicConsumer.acknowledge(currentTask.getMessageId()); + currentTask = null; + } else { + return retval; + } + } + } + + private void getBatchSourceConfigs(Map config) { + if (!config.containsKey(BatchSourceConfig.BATCHSOURCE_CONFIG_KEY) + || !config.containsKey(BatchSourceConfig.BATCHSOURCE_CLASSNAME_KEY)) { + throw new IllegalArgumentException("Batch Configs cannot be found"); + } + + String batchSourceConfigJson = (String) config.get(BatchSourceConfig.BATCHSOURCE_CONFIG_KEY); + this.batchSourceConfig = new Gson().fromJson(batchSourceConfigJson, BatchSourceConfig.class); + this.batchSourceClassName = (String)config.get(BatchSourceConfig.BATCHSOURCE_CLASSNAME_KEY); + } + + private void initializeBatchSource() { + // First init the batchsource + ClassLoader clsLoader = Thread.currentThread().getContextClassLoader(); + Object userClassObject = Reflections.createInstance( + batchSourceClassName, + clsLoader); + if (userClassObject instanceof BatchSource) { + batchSource = (BatchSource) userClassObject; + } else { + throw new IllegalArgumentException("BatchSource does not implement the correct interface"); + } + + // next init the discovery triggerer + Object discoveryClassObject = Reflections.createInstance( + batchSourceConfig.getDiscoveryTriggererClassName(), + clsLoader); + if (discoveryClassObject instanceof BatchSourceTriggerer) { + discoveryTriggerer = (BatchSourceTriggerer) discoveryClassObject; + } else { + throw new IllegalArgumentException("BatchSourceTriggerer does not implement the correct interface"); + } + } + + private void start() throws Exception { + // This is the first thing to do to ensure that any tasks discovered during the discover + // phase are not lost + setupInstanceSubscription(); + if (sourceContext.getInstanceId() == 0) { + discoveryTriggerer.init(batchSourceConfig.getDiscoveryTriggererConfig(), + this.sourceContext); + discoveryTriggerer.start(this::triggerDiscover); + } + batchSource.open(this.config, this.sourceContext); + } + + private void triggerDiscover(String discoveredEvent) { + try { + batchSource.discover((task) -> this.taskEater(discoveredEvent, task)); + } catch (Exception e) { + log.error("Error on discover", e); + throw new RuntimeException(e); + } + } + + private void taskEater(String discoveredEvent, byte[] task) { + try { + Map properties = new HashMap<>(); + properties.put("discoveredEvent", discoveredEvent); + properties.put("produceTime", String.valueOf(System.currentTimeMillis())); + TypedMessageBuilder message = sourceContext.newOutputMessage(intermediateTopicName, Schema.BYTES); + message.value(task).properties(properties); + message.send(); + } catch (Exception e) { + log.error("error writing discovered task to intermediate topic", e); + throw new RuntimeException("error writing discovered task to intermediate topic"); + } + } + + private void prepareInternal() { + try { + batchSource.prepare(currentTask.getValue()); + } catch (Exception e) { + log.error("Error on prepare", e); + throw new RuntimeException(e); + } + } + + public org.apache.pulsar.functions.api.Record readInternal() { + try { + Record record = batchSource.readNext(); + log.info("Record: {}", record); + if (record != null) { + return record; + } + } catch (Exception e) { + log.error("Error on read", e); + throw new RuntimeException(e); + } + return null; + } + + @Override + public void close() throws Exception { + this.stop(); + } + + private void stop() throws Exception { + if (discoveryTriggerer != null) { + discoveryTriggerer.stop(); + discoveryTriggerer = null; + } + if (intermediateTopicConsumer != null) { + intermediateTopicConsumer.close(); + intermediateTopicConsumer = null; + } + } + + private void setupInstanceSubscription() { + String subName = SourceConfigUtils.computeBatchSourceInstanceSubscriptionName( + sourceContext.getTenant(), sourceContext.getNamespace(), + sourceContext.getSourceName()); + try { + Actions.newBuilder() + .addAction( + Actions.Action.builder() + .actionName(String.format("Setting up instance consumer for BatchSource intermediate " + + "topic for function %s", FunctionCommon.getFullyQualifiedName( + sourceContext.getTenant(), sourceContext.getNamespace(), + sourceContext.getSourceName()))) + .numRetries(10) + .sleepBetweenInvocationsMs(1000) + .supplier(() -> { + try { + CompletableFuture> cf = sourceContext.newConsumerBuilder(Schema.BYTES) + .subscriptionName(subName) + .subscriptionType(SubscriptionType.Shared) + .topic(intermediateTopicName) + .subscribeAsync(); + intermediateTopicConsumer = cf.join(); + return Actions.ActionResult.builder() + .success(true) + .build(); + } catch (Exception e) { + return Actions.ActionResult.builder() + .success(false) + .build(); + } + }) + .build()) + .run(); + } catch (InterruptedException e) { + log.error("Error setting up instance subscription for intermediate topic", e); + throw new RuntimeException(e); + } + } + + private void retrieveNextTask() throws Exception { + currentTask = intermediateTopicConsumer.receive(); + return; + } + +} + diff --git a/pulsar-io/batch/src/test/java/org/apache/pulsar/io/batch/BatchSourceExecutorTest.java b/pulsar-io/batch/src/test/java/org/apache/pulsar/io/batch/BatchSourceExecutorTest.java new file mode 100644 index 0000000000000..d9d4112b74d5e --- /dev/null +++ b/pulsar-io/batch/src/test/java/org/apache/pulsar/io/batch/BatchSourceExecutorTest.java @@ -0,0 +1,249 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.io.batch; + + +import com.google.gson.Gson; +import lombok.Getter; +import org.apache.pulsar.client.api.*; +import org.apache.pulsar.common.io.BatchSourceConfig; +import org.apache.pulsar.functions.api.Record; + +import org.apache.pulsar.io.core.BatchSource; +import org.apache.pulsar.io.core.BatchSourceTriggerer; +import org.apache.pulsar.io.core.SourceContext; +import org.mockito.Mockito; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; +import org.testng.Assert; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +import java.util.*; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CyclicBarrier; +import java.util.function.Consumer; + +/** + * Unit tests for {@link org.apache.pulsar.io.batch.BatchSourceExecutor} + */ +public class BatchSourceExecutorTest { + + public static class TestBatchSource implements BatchSource { + @Getter + private static int prepareCount; + @Getter + private static int discoverCount; + @Getter + private static int recordCount; + private Record record = Mockito.mock(Record.class); + public TestBatchSource() { } + + @Override + public void open(Map config, SourceContext context) throws Exception { + if (!config.containsKey("foo")) { + throw new IllegalArgumentException("Bad config passed to TestBatchSource"); + } + } + + @Override + public void discover(Consumer taskEater) throws Exception { + byte[] retval = new byte[10]; + discoverCount++; + taskEater.accept(retval); + } + + @Override + public void prepare(byte[] task) throws Exception { + prepareCount++; + } + + @Override + public Record readNext() throws Exception { + if (++recordCount % 5 == 0) { + return null; + } else { + return record; + } + } + + @Override + public void close() throws Exception { + + } + } + + public static class TestDiscoveryTriggerer implements BatchSourceTriggerer { + private Consumer trigger; + private Thread thread; + + public TestDiscoveryTriggerer() { } + + @Override + public void init(Map config, SourceContext sourceContext) throws Exception { + if (!config.containsKey("DELAY_MS")) { + throw new IllegalArgumentException("Bad config passed to TestTriggerer"); + } + } + + @Override + public void start(Consumer trigger) { + this.trigger = trigger; + thread = new Thread(() -> { + while(true) { + try { + Thread.sleep(100); + trigger.accept("Triggered"); + } catch (InterruptedException e) { + break; + } + } + }); + thread.start(); + } + + @Override + public void stop() { + thread.interrupt(); + try { + thread.join(); + } catch (Exception e) { + } + } + } + + private TestBatchSource testBatchSource; + private BatchSourceConfig testBatchConfig; + private Map config; + private BatchSourceExecutor batchSourceExecutor; + private SourceContext context; + private ConsumerBuilder consumerBuilder; + private org.apache.pulsar.client.api.Consumer consumer; + private TypedMessageBuilder messageBuilder; + private CyclicBarrier discoveryBarrier; + private Message discoveredTask; + + @BeforeMethod + public void setUp() throws Exception { + testBatchSource = new TestBatchSource(); + batchSourceExecutor = new BatchSourceExecutor<>(); + context = Mockito.mock(SourceContext.class); + config = new HashMap<>(); + config.put("foo", "bar"); + testBatchConfig = new BatchSourceConfig(); + testBatchConfig.setDiscoveryTriggererClassName(TestDiscoveryTriggerer.class.getName()); + Map triggererConfig = new HashMap<>(); + triggererConfig.put("DELAY_MS", 500); + testBatchConfig.setDiscoveryTriggererConfig(triggererConfig); + config.put(BatchSourceConfig.BATCHSOURCE_CONFIG_KEY, new Gson().toJson(testBatchConfig)); + config.put(BatchSourceConfig.BATCHSOURCE_CLASSNAME_KEY, TestBatchSource.class.getName()); + Mockito.doReturn("test-function").when(context).getSourceName(); + Mockito.doReturn("test-namespace").when(context).getNamespace(); + Mockito.doReturn("test-tenant").when(context).getTenant(); + Mockito.doReturn(0).when(context).getInstanceId(); + consumerBuilder = Mockito.mock(ConsumerBuilder.class); + Mockito.doReturn(consumerBuilder).when(consumerBuilder).subscriptionName(Mockito.any()); + Mockito.doReturn(consumerBuilder).when(consumerBuilder).subscriptionType(Mockito.any()); + Mockito.doReturn(consumerBuilder).when(consumerBuilder).topic(Mockito.any()); + discoveredTask = Mockito.mock(Message.class); + consumer = Mockito.mock(org.apache.pulsar.client.api.Consumer.class); + Mockito.doReturn(discoveredTask).when(consumer).receive(); + Mockito.doReturn(CompletableFuture.completedFuture(consumer)).when(consumerBuilder).subscribeAsync(); + Mockito.doReturn(consumerBuilder).when(context).newConsumerBuilder(Schema.BYTES); + messageBuilder = Mockito.mock(TypedMessageBuilder.class); + Mockito.doReturn(messageBuilder).when(messageBuilder).value(Mockito.any()); + Mockito.doReturn(messageBuilder).when(messageBuilder).properties(Mockito.any()); + Mockito.doReturn(messageBuilder).when(context).newOutputMessage(Mockito.anyString(), Mockito.any()); + + // Discovery + discoveryBarrier = new CyclicBarrier(2); + Mockito.doAnswer(new Answer() { + @Override public MessageId answer(InvocationOnMock invocation) { + try { + discoveryBarrier.await(); + } catch (Exception e) { + throw new RuntimeException(); + } + return null; + } + }).when(messageBuilder).send(); + } + + @AfterMethod + public void cleanUp() { } + + @Test(expectedExceptions = IllegalArgumentException.class, expectedExceptionsMessageRegExp = "Batch Configs cannot be found") + public void testWithoutRightConfig() throws Exception { + config.clear(); + batchSourceExecutor.open(config, context); + } + + @Test(expectedExceptions = IllegalArgumentException.class, expectedExceptionsMessageRegExp = "BatchSourceTriggerer does not implement the correct interface") + public void testWithoutRightTriggerer() throws Exception { + testBatchConfig.setDiscoveryTriggererClassName(TestBatchSource.class.getName()); + config.put(BatchSourceConfig.BATCHSOURCE_CONFIG_KEY, new Gson().toJson(testBatchConfig)); + batchSourceExecutor.open(config, context); + } + + @Test(expectedExceptions = IllegalArgumentException.class, expectedExceptionsMessageRegExp = "Bad config passed to TestTriggerer") + public void testWithoutRightTriggererConfig() throws Exception { + Map badConfig = new HashMap<>(); + badConfig.put("something", "else"); + testBatchConfig.setDiscoveryTriggererConfig(badConfig); + config.put(BatchSourceConfig.BATCHSOURCE_CONFIG_KEY, new Gson().toJson(testBatchConfig)); + batchSourceExecutor.open(config, context); + } + + @Test(expectedExceptions = IllegalArgumentException.class, expectedExceptionsMessageRegExp = "BatchSource does not implement the correct interface") + public void testWithoutRightSource() throws Exception { + config.put(BatchSourceConfig.BATCHSOURCE_CLASSNAME_KEY, TestDiscoveryTriggerer.class.getName()); + batchSourceExecutor.open(config, context); + } + + @Test(expectedExceptions = IllegalArgumentException.class, expectedExceptionsMessageRegExp = "Bad config passed to TestBatchSource") + public void testWithoutRightSourceConfig() throws Exception { + config.remove("foo"); + config.put("something", "else"); + batchSourceExecutor.open(config, context); + } + + @Test + public void testOpenWithRightSource() throws Exception { + batchSourceExecutor.open(config, context); + } + + @Test + public void testLifeCycle() throws Exception { + batchSourceExecutor.open(config, context); + Assert.assertTrue(testBatchSource.getDiscoverCount() < 1); + discoveryBarrier.await(); + Assert.assertTrue(testBatchSource.getDiscoverCount() >= 1); + Assert.assertTrue(testBatchSource.getDiscoverCount() <= 2); + for (int i = 0; i < 5; ++i) { + batchSourceExecutor.read(); + } + Assert.assertEquals(testBatchSource.getRecordCount(), 6); + Assert.assertTrue(testBatchSource.getDiscoverCount() >= 1); + Assert.assertTrue(testBatchSource.getDiscoverCount() <= 2); + discoveryBarrier.await(); + Assert.assertTrue(testBatchSource.getDiscoverCount() >= 2); + Assert.assertTrue(testBatchSource.getDiscoverCount() <= 3); + } +} diff --git a/pulsar-io/core/src/main/java/org/apache/pulsar/io/core/BatchSource.java b/pulsar-io/core/src/main/java/org/apache/pulsar/io/core/BatchSource.java new file mode 100644 index 0000000000000..285f3c70473be --- /dev/null +++ b/pulsar-io/core/src/main/java/org/apache/pulsar/io/core/BatchSource.java @@ -0,0 +1,84 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.io.core; + + +import org.apache.pulsar.functions.api.Record; + +import java.util.Map; +import java.util.function.Consumer; + +/** + * Interface for writing Batch sources + * + * The lifecycle of the BatchSource is as follows + * 1. open - called once when connector is started. Can use method to perform + * certain one-time operations such as init/setup operations. This is called on all + * instances of the source and is analogous to the open method of the streaming Source api. + * 2. discover (called only on one instance (Currently instance zero(0), but might change later)) + * - The discovery phase will be executed on one instance of the connector. + * - discover is triggered by the BatchSourceTriggerer class configured for this source. + * - As and when discover discovers new tasks, it will emit them using the taskEater method. + * - The framework will distribute the discovered tasks among all instances + * 3. prepare - is called on an instance when there is a new discovered task assigned for that instance + * - The framework decides which discovered task is routed to which source instance. The connector + * does not currently have a way to influence this. + * - prepare is only called when the instance has fetched all records using readNext for its previously + * assigned discovered task. + * 4. readNext is called repeatedly by the framework to fetch the next record. If there are no + * more records available to emit, the connector should return null. That indicates + * that all records for that particular discovered task is complete. + * 5. close is called when the source is stopped/deleted. This is analogous to the streaming Source api. + * + */ + +public interface BatchSource extends AutoCloseable { + + /** + * Open connector with configuration. + * + * @param config config that's supplied for source + * @param context environment where the source connector is running + * @throws Exception IO type exceptions when opening a connector + */ + void open(final Map config, SourceContext context) throws Exception; + + /** + * Discovery phase of a connector. This phase will only be run on one instance, i.e. instance 0, of the connector. + * Implementations use the taskEater consumer to output serialized representation of tasks as they are discovered. + * + * @param taskEater function to notify the framework about the new task received. + * @throws Exception during discover + */ + void discover(Consumer taskEater) throws Exception; + + /** + * Called when a new task appears for this connector instance. + * + * @param task the serialized representation of the task + */ + void prepare(byte[] task) throws Exception; + + /** + * Read data and return a record + * Return null if no more records are present for this task + * @return a record + */ + Record readNext() throws Exception; +} \ No newline at end of file diff --git a/pulsar-io/core/src/main/java/org/apache/pulsar/io/core/BatchSourceTriggerer.java b/pulsar-io/core/src/main/java/org/apache/pulsar/io/core/BatchSourceTriggerer.java new file mode 100644 index 0000000000000..8f038158e49c4 --- /dev/null +++ b/pulsar-io/core/src/main/java/org/apache/pulsar/io/core/BatchSourceTriggerer.java @@ -0,0 +1,71 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.io.core; + +import java.util.Map; +import java.util.function.Consumer; + +/** + * This is an interface for defining BatchSource triggerers. These triggerers trigger + * the discovery method of the batch source that they are attached to. A BatchSource + * is configured to use a particular BatchSource Triggerer at the time of job submission. + * + * The interface and lifecycle is as follows + * 1. init - Called after the object is created by reflection. + * - The triggerer is created on only one instance of the source job. + * - The triggerer is passed its configuration in the init method. + * - Trigger also has access to the SourceContext of the BatchSource that it is attached to. + * It can use this context to get metadata information about the source as well things like secrets + * - This method just inits the triggerer. It doesn't start its execution. + * 2. start - Is called to actually start the running of the triggerer. + * - Triggerer will use the 'trigger' ConsumerFunction to actually trigger the discovery process + * 3. stop - Stop from further triggering discovers + * + */ + +public interface BatchSourceTriggerer { + + /** + * initializes the Triggerer with given config. Note that the triggerer doesn't start running + * until start is called. + * + * @param config config needed for triggerer to run + * @param sourceContext The source context associated with the source + * The parameter passed to this trigger function is an optional description of the event that caused the trigger + * @throws Exception throws any exceptions when initializing + */ + void init(Map config, SourceContext sourceContext) throws Exception; + + /** + * Triggerer should actually start looking out for trigger conditions. + * + * @param trigger The function to be called when its time to trigger the discover + * This function can be passed any metadata about this particular + * trigger event as its argument + * This method should return immediately. It is expected that implementations will use their own mechanisms + * to schedule the triggers. + */ + void start(Consumer trigger); + + /** + * Triggerer should stop triggering. + * + */ + void stop(); +} \ No newline at end of file diff --git a/pulsar-io/pom.xml b/pulsar-io/pom.xml index 2e875e4e85050..ccf80e091c6f9 100644 --- a/pulsar-io/pom.xml +++ b/pulsar-io/pom.xml @@ -33,6 +33,9 @@ core + batch + batch-discovery-triggerers + batch-data-generator common docs aws