From efa2473ab9ce1b1806484d3c41dd22cffeb8f6a3 Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Wed, 11 Nov 2020 17:17:51 +0800 Subject: [PATCH 1/9] Support key_based batch builder for functions and sources Signed-off-by: xiaolong.ran --- .../apache/pulsar/admin/cli/CmdFunctions.java | 6 ++++++ .../apache/pulsar/admin/cli/CmdSources.java | 7 +++++++ .../common/functions/FunctionConfig.java | 2 ++ .../common/functions/ProducerConfig.java | 1 + .../apache/pulsar/common/io/SourceConfig.java | 2 ++ .../pulsar/functions/sink/PulsarSink.java | 20 ++++++++----------- .../proto/src/main/proto/Function.proto | 1 + .../functions/utils/FunctionConfigUtils.java | 3 +++ .../functions/utils/SourceConfigUtils.java | 3 +++ .../utils/FunctionConfigUtilsTest.java | 1 + .../utils/SourceConfigUtilsTest.java | 1 + 11 files changed, 35 insertions(+), 12 deletions(-) diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java index dacc8394c499f..f9670e0bfb6ad 100644 --- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java +++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java @@ -266,6 +266,8 @@ abstract class FunctionDetailsCommand extends BaseCommand { protected Boolean DEPRECATED_retainOrdering; @Parameter(names = "--retain-ordering", description = "Function consumes and processes messages in order") protected Boolean retainOrdering; + @Parameter(names = "--batch-builder", description = "BatcherBuilder provides two types of batch construction methods, DEFAULT and KEY_BASED. The default value is: DEFAULT") + protected String batchBuilder; @Parameter(names = "--forward-source-message-property", description = "Forwarding input message's properties to output topic when processing") protected Boolean forwardSourceMessageProperty = true; @Parameter(names = "--subs-name", description = "Pulsar source subscription name if user wants a specific subscription-name for input-topic consumer") @@ -419,6 +421,10 @@ void processArguments() throws Exception { functionConfig.setRetainOrdering(retainOrdering); } + if (isNotBlank(batchBuilder)) { + functionConfig.setBatchBuilder(batchBuilder); + } + if (null != forwardSourceMessageProperty) { functionConfig.setForwardSourceMessageProperty(forwardSourceMessageProperty); } diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java index fcc580ad68ba1..521924b4bc821 100644 --- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java +++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java @@ -274,6 +274,9 @@ abstract class SourceDetailsCommand extends BaseCommand { @Parameter(names = "--producer-config", description = "The custom producer configuration (as a JSON string)") protected String producerConfig; + @Parameter(names = "--batch-builder", description = "BatchBuilder provides two types of batch construction methods, DEFAULT and KEY_BASED. The default value is: DEFAULT") + protected String batchBuilder; + @Parameter(names = "--deserializationClassName", description = "The SerDe classname for the source", hidden = true) protected String DEPRECATED_deserializationClassName; @Parameter(names = "--deserialization-classname", description = "The SerDe classname for the source") @@ -360,6 +363,10 @@ void processArguments() throws Exception { sourceConfig.setSchemaType(schemaType); } + if (null != batchBuilder) { + sourceConfig.setBatchBuilder(batchBuilder); + } + if (null != processingGuarantees) { sourceConfig.setProcessingGuarantees(processingGuarantees); } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/functions/FunctionConfig.java b/pulsar-common/src/main/java/org/apache/pulsar/common/functions/FunctionConfig.java index 9cd8ef6f9b5f4..c9f4645ed774a 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/functions/FunctionConfig.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/functions/FunctionConfig.java @@ -95,6 +95,8 @@ public enum Runtime { private Boolean retainOrdering; // Do we want the same function instance to process all data keyed by the input topic's message key private Boolean retainKeyOrdering; + // batchBuilder provides two types of batch construction methods, DEFAULT and KEY_BASED + private String batchBuilder; private Boolean forwardSourceMessageProperty; private Map userConfig; // This is a map of secretName(aka how the secret is going to be diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/functions/ProducerConfig.java b/pulsar-common/src/main/java/org/apache/pulsar/common/functions/ProducerConfig.java index 5b686638dbf7b..0d2f4b48b19be 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/functions/ProducerConfig.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/functions/ProducerConfig.java @@ -37,4 +37,5 @@ public class ProducerConfig { private Integer maxPendingMessagesAcrossPartitions; private Boolean useThreadLocalProducers; private CryptoConfig cryptoConfig; + private String batchBuilder; } 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 31a8634a2dcf7..79a5f81f3b9cf 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 @@ -69,4 +69,6 @@ public class SourceConfig { // If this is a BatchSource, its batch related configs are stored here private BatchSourceConfig batchSourceConfig; + // batchBuilder provides two types of batch construction methods, DEFAULT and KEY_BASED + private String batchBuilder; } diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java index f1219501f6830..f94b30bac529d 100644 --- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java +++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java @@ -24,18 +24,7 @@ import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.reflect.ConstructorUtils; -import org.apache.pulsar.client.api.CompressionType; -import org.apache.pulsar.client.api.CryptoKeyReader; -import org.apache.pulsar.client.api.HashingScheme; -import org.apache.pulsar.client.api.MessageId; -import org.apache.pulsar.client.api.MessageRoutingMode; -import org.apache.pulsar.client.api.Producer; -import org.apache.pulsar.client.api.ProducerBuilder; -import org.apache.pulsar.client.api.ProducerCryptoFailureAction; -import org.apache.pulsar.client.api.PulsarClient; -import org.apache.pulsar.client.api.PulsarClientException; -import org.apache.pulsar.client.api.Schema; -import org.apache.pulsar.client.api.TypedMessageBuilder; +import org.apache.pulsar.client.api.*; import org.apache.pulsar.client.impl.schema.KeyValueSchema; import org.apache.pulsar.common.functions.ConsumerConfig; import org.apache.pulsar.common.functions.CryptoConfig; @@ -133,6 +122,13 @@ public Producer createProducer(PulsarClient client, String topic, String prod builder.addEncryptionKey(encryptionKeyName); } } + if (producerConfig.getBatchBuilder() != null) { + if (producerConfig.getBatchBuilder().equals("KEY_BASED")) { + builder.batcherBuilder(BatcherBuilder.KEY_BASED); + } else { + builder.batcherBuilder(BatcherBuilder.DEFAULT); + } + } } return builder.properties(properties).create(); } diff --git a/pulsar-functions/proto/src/main/proto/Function.proto b/pulsar-functions/proto/src/main/proto/Function.proto index 4243fb622272c..4b840a3daadfd 100644 --- a/pulsar-functions/proto/src/main/proto/Function.proto +++ b/pulsar-functions/proto/src/main/proto/Function.proto @@ -108,6 +108,7 @@ message ProducerSpec { int32 maxPendingMessagesAcrossPartitions = 2; bool useThreadLocalProducers = 3; CryptoSpec cryptoSpec = 4; + string batchBuilder = 5; } message CryptoSpec { diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java index bdf250a60bfeb..fc23cfecbc49a 100644 --- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java +++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java @@ -226,6 +226,9 @@ public static FunctionDetails convert(FunctionConfig functionConfig, ClassLoader if (producerConf.getCryptoConfig() != null) { pbldr.setCryptoSpec(CryptoUtils.convert(producerConf.getCryptoConfig())); } + if (producerConf.getBatchBuilder() != null) { + pbldr.setBatchBuilder(producerConf.getBatchBuilder()); + } sinkSpecBuilder.setProducerSpec(pbldr.build()); } functionDetailsBuilder.setSink(sinkSpecBuilder); 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 aec11fee898e5..83963d62afe90 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 @@ -163,6 +163,9 @@ public static FunctionDetails convert(SourceConfig sourceConfig, ExtractedSource if (conf.getCryptoConfig() != null) { pbldr.setCryptoSpec(CryptoUtils.convert(conf.getCryptoConfig())); } + if (conf.getBatchBuilder() != null) { + pbldr.setBatchBuilder(conf.getBatchBuilder()); + } sinkSpecBuilder.setProducerSpec(pbldr.build()); } diff --git a/pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/FunctionConfigUtilsTest.java b/pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/FunctionConfigUtilsTest.java index fcafd49ab9e1c..4f8feda58b549 100644 --- a/pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/FunctionConfigUtilsTest.java +++ b/pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/FunctionConfigUtilsTest.java @@ -449,6 +449,7 @@ private FunctionConfig createFunctionConfig() { functionConfig.setRetainOrdering(false); functionConfig.setRetainKeyOrdering(false); functionConfig.setSubscriptionPosition(SubscriptionInitialPosition.Earliest); + functionConfig.setBatchBuilder("DEFAULT"); functionConfig.setForwardSourceMessageProperty(false); functionConfig.setUserConfig(new HashMap<>()); functionConfig.setAutoAck(true); 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 520b416e722e4..ace7842f79517 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 @@ -341,6 +341,7 @@ private SourceConfig createSourceConfig() { sourceConfig.setSerdeClassName("test-serde"); sourceConfig.setParallelism(1); sourceConfig.setRuntimeFlags("-DKerberos"); + sourceConfig.setBatchBuilder("DEFAULT"); sourceConfig.setProcessingGuarantees(FunctionConfig.ProcessingGuarantees.ATLEAST_ONCE); Map consumerConfigs = new HashMap<>(); From 9b93663761753982ecf1b77180cba31e76fc109a Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Wed, 11 Nov 2020 17:30:35 +0800 Subject: [PATCH 2/9] Avoid using * in import Signed-off-by: xiaolong.ran --- .../apache/pulsar/functions/sink/PulsarSink.java | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java index f94b30bac529d..9ba40591df2ad 100644 --- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java +++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java @@ -24,7 +24,19 @@ import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.reflect.ConstructorUtils; -import org.apache.pulsar.client.api.*; +import org.apache.pulsar.client.api.BatcherBuilder; +import org.apache.pulsar.client.api.CompressionType; +import org.apache.pulsar.client.api.CryptoKeyReader; +import org.apache.pulsar.client.api.HashingScheme; +import org.apache.pulsar.client.api.MessageId; +import org.apache.pulsar.client.api.MessageRoutingMode; +import org.apache.pulsar.client.api.Producer; +import org.apache.pulsar.client.api.ProducerBuilder; +import org.apache.pulsar.client.api.ProducerCryptoFailureAction; +import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.TypedMessageBuilder; import org.apache.pulsar.client.impl.schema.KeyValueSchema; import org.apache.pulsar.common.functions.ConsumerConfig; import org.apache.pulsar.common.functions.CryptoConfig; From ac089ea17ec11996d8627d894065899abb4d9831 Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Thu, 12 Nov 2020 15:15:51 +0800 Subject: [PATCH 3/9] fix ci error Signed-off-by: xiaolong.ran --- .../org/apache/pulsar/functions/utils/SourceConfigUtils.java | 3 +++ .../apache/pulsar/functions/utils/SourceConfigUtilsTest.java | 2 +- 2 files changed, 4 insertions(+), 1 deletion(-) 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 83963d62afe90..04f524a3c0807 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 @@ -250,6 +250,9 @@ public static SourceConfig convertFromDetails(FunctionDetails functionDetails) { if (spec.hasCryptoSpec()) { producerConfig.setCryptoConfig(CryptoUtils.convertFromSpec(spec.getCryptoSpec())); } + if (spec.getBatchBuilder() != null) { + producerConfig.setBatchBuilder(spec.getBatchBuilder()); + } producerConfig.setUseThreadLocalProducers(spec.getUseThreadLocalProducers()); sourceConfig.setProducerConfig(producerConfig); } 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 ace7842f79517..bcece84fd9a88 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 @@ -341,7 +341,6 @@ private SourceConfig createSourceConfig() { sourceConfig.setSerdeClassName("test-serde"); sourceConfig.setParallelism(1); sourceConfig.setRuntimeFlags("-DKerberos"); - sourceConfig.setBatchBuilder("DEFAULT"); sourceConfig.setProcessingGuarantees(FunctionConfig.ProcessingGuarantees.ATLEAST_ONCE); Map consumerConfigs = new HashMap<>(); @@ -355,6 +354,7 @@ private SourceConfig createSourceConfig() { producerConfig.setMaxPendingMessages(100); producerConfig.setMaxPendingMessagesAcrossPartitions(1000); producerConfig.setUseThreadLocalProducers(true); + producerConfig.setBatchBuilder("DEFAULT"); sourceConfig.setProducerConfig(producerConfig); sourceConfig.setConfigs(configs); From 34a8a6a68f045680dd20e0ccf348cb3587b029c9 Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Thu, 12 Nov 2020 15:16:59 +0800 Subject: [PATCH 4/9] Support key_based feature for python function Signed-off-by: xiaolong.ran --- .../instance/src/main/python/Function_pb2.py | 234 ++++++++++++++---- .../src/main/python/python_instance.py | 8 + 2 files changed, 194 insertions(+), 48 deletions(-) diff --git a/pulsar-functions/instance/src/main/python/Function_pb2.py b/pulsar-functions/instance/src/main/python/Function_pb2.py index 3ee58e920fa3a..14a33a1255d2a 100644 --- a/pulsar-functions/instance/src/main/python/Function_pb2.py +++ b/pulsar-functions/instance/src/main/python/Function_pb2.py @@ -38,7 +38,7 @@ syntax='proto3', serialized_options=b'\n!org.apache.pulsar.functions.protoB\010Function', create_key=_descriptor._internal_create_key, - serialized_pb=b'\n\x0e\x46unction.proto\x12\x05proto\"3\n\tResources\x12\x0b\n\x03\x63pu\x18\x01 \x01(\x01\x12\x0b\n\x03ram\x18\x02 \x01(\x03\x12\x0c\n\x04\x64isk\x18\x03 \x01(\x03\"B\n\x0cRetryDetails\x12\x19\n\x11maxMessageRetries\x18\x01 \x01(\x05\x12\x17\n\x0f\x64\x65\x61\x64LetterTopic\x18\x02 \x01(\t\"\xe7\x05\n\x0f\x46unctionDetails\x12\x0e\n\x06tenant\x18\x01 \x01(\t\x12\x11\n\tnamespace\x18\x02 \x01(\t\x12\x0c\n\x04name\x18\x03 \x01(\t\x12\x11\n\tclassName\x18\x04 \x01(\t\x12\x10\n\x08logTopic\x18\x05 \x01(\t\x12\x39\n\x14processingGuarantees\x18\x06 \x01(\x0e\x32\x1b.proto.ProcessingGuarantees\x12\x12\n\nuserConfig\x18\x07 \x01(\t\x12\x12\n\nsecretsMap\x18\x10 \x01(\t\x12/\n\x07runtime\x18\x08 \x01(\x0e\x32\x1e.proto.FunctionDetails.Runtime\x12\x0f\n\x07\x61utoAck\x18\t \x01(\x08\x12\x13\n\x0bparallelism\x18\n \x01(\x05\x12!\n\x06source\x18\x0b \x01(\x0b\x32\x11.proto.SourceSpec\x12\x1d\n\x04sink\x18\x0c \x01(\x0b\x32\x0f.proto.SinkSpec\x12#\n\tresources\x18\r \x01(\x0b\x32\x10.proto.Resources\x12\x12\n\npackageUrl\x18\x0e \x01(\t\x12)\n\x0cretryDetails\x18\x0f \x01(\x0b\x32\x13.proto.RetryDetails\x12\x14\n\x0cruntimeFlags\x18\x11 \x01(\t\x12;\n\rcomponentType\x18\x12 \x01(\x0e\x32$.proto.FunctionDetails.ComponentType\x12\x1c\n\x14\x63ustomRuntimeOptions\x18\x13 \x01(\t\x12\x0f\n\x07\x62uiltin\x18\x14 \x01(\t\x12\x16\n\x0eretainOrdering\x18\x15 \x01(\x08\x12\x19\n\x11retainKeyOrdering\x18\x16 \x01(\x08\"\'\n\x07Runtime\x12\x08\n\x04JAVA\x10\x00\x12\n\n\x06PYTHON\x10\x01\x12\x06\n\x02GO\x10\x03\"@\n\rComponentType\x12\x0b\n\x07UNKNOWN\x10\x00\x12\x0c\n\x08\x46UNCTION\x10\x01\x12\n\n\x06SOURCE\x10\x02\x12\x08\n\x04SINK\x10\x03\"\xba\x03\n\x0c\x43onsumerSpec\x12\x12\n\nschemaType\x18\x01 \x01(\t\x12\x16\n\x0eserdeClassName\x18\x02 \x01(\t\x12\x16\n\x0eisRegexPattern\x18\x03 \x01(\x08\x12@\n\x11receiverQueueSize\x18\x04 \x01(\x0b\x32%.proto.ConsumerSpec.ReceiverQueueSize\x12\x43\n\x10schemaProperties\x18\x05 \x03(\x0b\x32).proto.ConsumerSpec.SchemaPropertiesEntry\x12G\n\x12\x63onsumerProperties\x18\x06 \x03(\x0b\x32+.proto.ConsumerSpec.ConsumerPropertiesEntry\x1a\"\n\x11ReceiverQueueSize\x12\r\n\x05value\x18\x01 \x01(\x05\x1a\x37\n\x15SchemaPropertiesEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\x1a\x39\n\x17\x43onsumerPropertiesEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\"w\n\x0cProducerSpec\x12\x1a\n\x12maxPendingMessages\x18\x01 \x01(\x05\x12*\n\"maxPendingMessagesAcrossPartitions\x18\x02 \x01(\x05\x12\x1f\n\x17useThreadLocalProducers\x18\x03 \x01(\x08\"\xe2\x04\n\nSourceSpec\x12\x11\n\tclassName\x18\x01 \x01(\t\x12\x0f\n\x07\x63onfigs\x18\x02 \x01(\t\x12\x15\n\rtypeClassName\x18\x05 \x01(\t\x12\x31\n\x10subscriptionType\x18\x03 \x01(\x0e\x32\x17.proto.SubscriptionType\x12Q\n\x16topicsToSerDeClassName\x18\x04 \x03(\x0b\x32-.proto.SourceSpec.TopicsToSerDeClassNameEntryB\x02\x18\x01\x12\x35\n\ninputSpecs\x18\n \x03(\x0b\x32!.proto.SourceSpec.InputSpecsEntry\x12\x11\n\ttimeoutMs\x18\x06 \x01(\x04\x12\x19\n\rtopicsPattern\x18\x07 \x01(\tB\x02\x18\x01\x12\x0f\n\x07\x62uiltin\x18\x08 \x01(\t\x12\x18\n\x10subscriptionName\x18\t \x01(\t\x12\x1b\n\x13\x63leanupSubscription\x18\x0b \x01(\x08\x12\x39\n\x14subscriptionPosition\x18\x0c \x01(\x0e\x32\x1b.proto.SubscriptionPosition\x12$\n\x1cnegativeAckRedeliveryDelayMs\x18\r \x01(\x04\x1a=\n\x1bTopicsToSerDeClassNameEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\x1a\x46\n\x0fInputSpecsEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\"\n\x05value\x18\x02 \x01(\x0b\x32\x13.proto.ConsumerSpec:\x02\x38\x01\"\xdc\x03\n\x08SinkSpec\x12\x11\n\tclassName\x18\x01 \x01(\t\x12\x0f\n\x07\x63onfigs\x18\x02 \x01(\t\x12\x15\n\rtypeClassName\x18\x05 \x01(\t\x12\r\n\x05topic\x18\x03 \x01(\t\x12)\n\x0cproducerSpec\x18\x0b \x01(\x0b\x32\x13.proto.ProducerSpec\x12\x16\n\x0eserDeClassName\x18\x04 \x01(\t\x12\x0f\n\x07\x62uiltin\x18\x06 \x01(\t\x12\x12\n\nschemaType\x18\x07 \x01(\t\x12$\n\x1c\x66orwardSourceMessageProperty\x18\x08 \x01(\x08\x12?\n\x10schemaProperties\x18\t \x03(\x0b\x32%.proto.SinkSpec.SchemaPropertiesEntry\x12\x43\n\x12\x63onsumerProperties\x18\n \x03(\x0b\x32\'.proto.SinkSpec.ConsumerPropertiesEntry\x1a\x37\n\x15SchemaPropertiesEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\x1a\x39\n\x17\x43onsumerPropertiesEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\"H\n\x17PackageLocationMetaData\x12\x13\n\x0bpackagePath\x18\x01 \x01(\t\x12\x18\n\x10originalFileName\x18\x02 \x01(\t\"\xf0\x02\n\x10\x46unctionMetaData\x12/\n\x0f\x66unctionDetails\x18\x01 \x01(\x0b\x32\x16.proto.FunctionDetails\x12\x37\n\x0fpackageLocation\x18\x02 \x01(\x0b\x32\x1e.proto.PackageLocationMetaData\x12\x0f\n\x07version\x18\x03 \x01(\x04\x12\x12\n\ncreateTime\x18\x04 \x01(\x04\x12\x43\n\x0einstanceStates\x18\x05 \x03(\x0b\x32+.proto.FunctionMetaData.InstanceStatesEntry\x12;\n\x10\x66unctionAuthSpec\x18\x06 \x01(\x0b\x32!.proto.FunctionAuthenticationSpec\x1aK\n\x13InstanceStatesEntry\x12\x0b\n\x03key\x18\x01 \x01(\x05\x12#\n\x05value\x18\x02 \x01(\x0e\x32\x14.proto.FunctionState:\x02\x38\x01\"<\n\x1a\x46unctionAuthenticationSpec\x12\x0c\n\x04\x64\x61ta\x18\x01 \x01(\x0c\x12\x10\n\x08provider\x18\x02 \x01(\t\"Q\n\x08Instance\x12\x31\n\x10\x66unctionMetaData\x18\x01 \x01(\x0b\x32\x17.proto.FunctionMetaData\x12\x12\n\ninstanceId\x18\x02 \x01(\x05\"A\n\nAssignment\x12!\n\x08instance\x18\x01 \x01(\x0b\x32\x0f.proto.Instance\x12\x10\n\x08workerId\x18\x02 \x01(\t*O\n\x14ProcessingGuarantees\x12\x10\n\x0c\x41TLEAST_ONCE\x10\x00\x12\x0f\n\x0b\x41TMOST_ONCE\x10\x01\x12\x14\n\x10\x45\x46\x46\x45\x43TIVELY_ONCE\x10\x02*<\n\x10SubscriptionType\x12\n\n\x06SHARED\x10\x00\x12\x0c\n\x08\x46\x41ILOVER\x10\x01\x12\x0e\n\nKEY_SHARED\x10\x02*0\n\x14SubscriptionPosition\x12\n\n\x06LATEST\x10\x00\x12\x0c\n\x08\x45\x41RLIEST\x10\x01*)\n\rFunctionState\x12\x0b\n\x07RUNNING\x10\x00\x12\x0b\n\x07STOPPED\x10\x01\x42-\n!org.apache.pulsar.functions.protoB\x08\x46unctionb\x06proto3' + serialized_pb=b'\n\x0e\x46unction.proto\x12\x05proto\"3\n\tResources\x12\x0b\n\x03\x63pu\x18\x01 \x01(\x01\x12\x0b\n\x03ram\x18\x02 \x01(\x03\x12\x0c\n\x04\x64isk\x18\x03 \x01(\x03\"B\n\x0cRetryDetails\x12\x19\n\x11maxMessageRetries\x18\x01 \x01(\x05\x12\x17\n\x0f\x64\x65\x61\x64LetterTopic\x18\x02 \x01(\t\"\xa2\x06\n\x0f\x46unctionDetails\x12\x0e\n\x06tenant\x18\x01 \x01(\t\x12\x11\n\tnamespace\x18\x02 \x01(\t\x12\x0c\n\x04name\x18\x03 \x01(\t\x12\x11\n\tclassName\x18\x04 \x01(\t\x12\x10\n\x08logTopic\x18\x05 \x01(\t\x12\x39\n\x14processingGuarantees\x18\x06 \x01(\x0e\x32\x1b.proto.ProcessingGuarantees\x12\x12\n\nuserConfig\x18\x07 \x01(\t\x12\x12\n\nsecretsMap\x18\x10 \x01(\t\x12/\n\x07runtime\x18\x08 \x01(\x0e\x32\x1e.proto.FunctionDetails.Runtime\x12\x0f\n\x07\x61utoAck\x18\t \x01(\x08\x12\x13\n\x0bparallelism\x18\n \x01(\x05\x12!\n\x06source\x18\x0b \x01(\x0b\x32\x11.proto.SourceSpec\x12\x1d\n\x04sink\x18\x0c \x01(\x0b\x32\x0f.proto.SinkSpec\x12#\n\tresources\x18\r \x01(\x0b\x32\x10.proto.Resources\x12\x12\n\npackageUrl\x18\x0e \x01(\t\x12)\n\x0cretryDetails\x18\x0f \x01(\x0b\x32\x13.proto.RetryDetails\x12\x14\n\x0cruntimeFlags\x18\x11 \x01(\t\x12;\n\rcomponentType\x18\x12 \x01(\x0e\x32$.proto.FunctionDetails.ComponentType\x12\x1c\n\x14\x63ustomRuntimeOptions\x18\x13 \x01(\t\x12\x0f\n\x07\x62uiltin\x18\x14 \x01(\t\x12\x16\n\x0eretainOrdering\x18\x15 \x01(\x08\x12\x19\n\x11retainKeyOrdering\x18\x16 \x01(\x08\x12\x39\n\x14subscriptionPosition\x18\x17 \x01(\x0e\x32\x1b.proto.SubscriptionPosition\"\'\n\x07Runtime\x12\x08\n\x04JAVA\x10\x00\x12\n\n\x06PYTHON\x10\x01\x12\x06\n\x02GO\x10\x03\"@\n\rComponentType\x12\x0b\n\x07UNKNOWN\x10\x00\x12\x0c\n\x08\x46UNCTION\x10\x01\x12\n\n\x06SOURCE\x10\x02\x12\x08\n\x04SINK\x10\x03\"\xe1\x03\n\x0c\x43onsumerSpec\x12\x12\n\nschemaType\x18\x01 \x01(\t\x12\x16\n\x0eserdeClassName\x18\x02 \x01(\t\x12\x16\n\x0eisRegexPattern\x18\x03 \x01(\x08\x12@\n\x11receiverQueueSize\x18\x04 \x01(\x0b\x32%.proto.ConsumerSpec.ReceiverQueueSize\x12\x43\n\x10schemaProperties\x18\x05 \x03(\x0b\x32).proto.ConsumerSpec.SchemaPropertiesEntry\x12G\n\x12\x63onsumerProperties\x18\x06 \x03(\x0b\x32+.proto.ConsumerSpec.ConsumerPropertiesEntry\x12%\n\ncryptoSpec\x18\x07 \x01(\x0b\x32\x11.proto.CryptoSpec\x1a\"\n\x11ReceiverQueueSize\x12\r\n\x05value\x18\x01 \x01(\x05\x1a\x37\n\x15SchemaPropertiesEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\x1a\x39\n\x17\x43onsumerPropertiesEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\"\xb4\x01\n\x0cProducerSpec\x12\x1a\n\x12maxPendingMessages\x18\x01 \x01(\x05\x12*\n\"maxPendingMessagesAcrossPartitions\x18\x02 \x01(\x05\x12\x1f\n\x17useThreadLocalProducers\x18\x03 \x01(\x08\x12%\n\ncryptoSpec\x18\x04 \x01(\x0b\x32\x11.proto.CryptoSpec\x12\x14\n\x0c\x62\x61tchBuilder\x18\x05 \x01(\t\"\xbb\x02\n\nCryptoSpec\x12 \n\x18\x63ryptoKeyReaderClassName\x18\x01 \x01(\t\x12\x1d\n\x15\x63ryptoKeyReaderConfig\x18\x02 \x01(\t\x12!\n\x19producerEncryptionKeyName\x18\x03 \x03(\t\x12\x44\n\x1bproducerCryptoFailureAction\x18\x04 \x01(\x0e\x32\x1f.proto.CryptoSpec.FailureAction\x12\x44\n\x1b\x63onsumerCryptoFailureAction\x18\x05 \x01(\x0e\x32\x1f.proto.CryptoSpec.FailureAction\"=\n\rFailureAction\x12\x08\n\x04\x46\x41IL\x10\x00\x12\x0b\n\x07\x44ISCARD\x10\x01\x12\x0b\n\x07\x43ONSUME\x10\x02\x12\x08\n\x04SEND\x10\n\"\xe2\x04\n\nSourceSpec\x12\x11\n\tclassName\x18\x01 \x01(\t\x12\x0f\n\x07\x63onfigs\x18\x02 \x01(\t\x12\x15\n\rtypeClassName\x18\x05 \x01(\t\x12\x31\n\x10subscriptionType\x18\x03 \x01(\x0e\x32\x17.proto.SubscriptionType\x12Q\n\x16topicsToSerDeClassName\x18\x04 \x03(\x0b\x32-.proto.SourceSpec.TopicsToSerDeClassNameEntryB\x02\x18\x01\x12\x35\n\ninputSpecs\x18\n \x03(\x0b\x32!.proto.SourceSpec.InputSpecsEntry\x12\x11\n\ttimeoutMs\x18\x06 \x01(\x04\x12\x19\n\rtopicsPattern\x18\x07 \x01(\tB\x02\x18\x01\x12\x0f\n\x07\x62uiltin\x18\x08 \x01(\t\x12\x18\n\x10subscriptionName\x18\t \x01(\t\x12\x1b\n\x13\x63leanupSubscription\x18\x0b \x01(\x08\x12\x39\n\x14subscriptionPosition\x18\x0c \x01(\x0e\x32\x1b.proto.SubscriptionPosition\x12$\n\x1cnegativeAckRedeliveryDelayMs\x18\r \x01(\x04\x1a=\n\x1bTopicsToSerDeClassNameEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\x1a\x46\n\x0fInputSpecsEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\"\n\x05value\x18\x02 \x01(\x0b\x32\x13.proto.ConsumerSpec:\x02\x38\x01\"\xdc\x03\n\x08SinkSpec\x12\x11\n\tclassName\x18\x01 \x01(\t\x12\x0f\n\x07\x63onfigs\x18\x02 \x01(\t\x12\x15\n\rtypeClassName\x18\x05 \x01(\t\x12\r\n\x05topic\x18\x03 \x01(\t\x12)\n\x0cproducerSpec\x18\x0b \x01(\x0b\x32\x13.proto.ProducerSpec\x12\x16\n\x0eserDeClassName\x18\x04 \x01(\t\x12\x0f\n\x07\x62uiltin\x18\x06 \x01(\t\x12\x12\n\nschemaType\x18\x07 \x01(\t\x12$\n\x1c\x66orwardSourceMessageProperty\x18\x08 \x01(\x08\x12?\n\x10schemaProperties\x18\t \x03(\x0b\x32%.proto.SinkSpec.SchemaPropertiesEntry\x12\x43\n\x12\x63onsumerProperties\x18\n \x03(\x0b\x32\'.proto.SinkSpec.ConsumerPropertiesEntry\x1a\x37\n\x15SchemaPropertiesEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\x1a\x39\n\x17\x43onsumerPropertiesEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\"H\n\x17PackageLocationMetaData\x12\x13\n\x0bpackagePath\x18\x01 \x01(\t\x12\x18\n\x10originalFileName\x18\x02 \x01(\t\"\xf0\x02\n\x10\x46unctionMetaData\x12/\n\x0f\x66unctionDetails\x18\x01 \x01(\x0b\x32\x16.proto.FunctionDetails\x12\x37\n\x0fpackageLocation\x18\x02 \x01(\x0b\x32\x1e.proto.PackageLocationMetaData\x12\x0f\n\x07version\x18\x03 \x01(\x04\x12\x12\n\ncreateTime\x18\x04 \x01(\x04\x12\x43\n\x0einstanceStates\x18\x05 \x03(\x0b\x32+.proto.FunctionMetaData.InstanceStatesEntry\x12;\n\x10\x66unctionAuthSpec\x18\x06 \x01(\x0b\x32!.proto.FunctionAuthenticationSpec\x1aK\n\x13InstanceStatesEntry\x12\x0b\n\x03key\x18\x01 \x01(\x05\x12#\n\x05value\x18\x02 \x01(\x0e\x32\x14.proto.FunctionState:\x02\x38\x01\"<\n\x1a\x46unctionAuthenticationSpec\x12\x0c\n\x04\x64\x61ta\x18\x01 \x01(\x0c\x12\x10\n\x08provider\x18\x02 \x01(\t\"Q\n\x08Instance\x12\x31\n\x10\x66unctionMetaData\x18\x01 \x01(\x0b\x32\x17.proto.FunctionMetaData\x12\x12\n\ninstanceId\x18\x02 \x01(\x05\"A\n\nAssignment\x12!\n\x08instance\x18\x01 \x01(\x0b\x32\x0f.proto.Instance\x12\x10\n\x08workerId\x18\x02 \x01(\t*O\n\x14ProcessingGuarantees\x12\x10\n\x0c\x41TLEAST_ONCE\x10\x00\x12\x0f\n\x0b\x41TMOST_ONCE\x10\x01\x12\x14\n\x10\x45\x46\x46\x45\x43TIVELY_ONCE\x10\x02*<\n\x10SubscriptionType\x12\n\n\x06SHARED\x10\x00\x12\x0c\n\x08\x46\x41ILOVER\x10\x01\x12\x0e\n\nKEY_SHARED\x10\x02*0\n\x14SubscriptionPosition\x12\n\n\x06LATEST\x10\x00\x12\x0c\n\x08\x45\x41RLIEST\x10\x01*)\n\rFunctionState\x12\x0b\n\x07RUNNING\x10\x00\x12\x0b\n\x07STOPPED\x10\x01\x42-\n!org.apache.pulsar.functions.protoB\x08\x46unctionb\x06proto3' ) _PROCESSINGGUARANTEES = _descriptor.EnumDescriptor( @@ -66,8 +66,8 @@ ], containing_type=None, serialized_options=None, - serialized_start=3207, - serialized_end=3286, + serialized_start=3685, + serialized_end=3764, ) _sym_db.RegisterEnumDescriptor(_PROCESSINGGUARANTEES) @@ -97,8 +97,8 @@ ], containing_type=None, serialized_options=None, - serialized_start=3288, - serialized_end=3348, + serialized_start=3766, + serialized_end=3826, ) _sym_db.RegisterEnumDescriptor(_SUBSCRIPTIONTYPE) @@ -123,8 +123,8 @@ ], containing_type=None, serialized_options=None, - serialized_start=3350, - serialized_end=3398, + serialized_start=3828, + serialized_end=3876, ) _sym_db.RegisterEnumDescriptor(_SUBSCRIPTIONPOSITION) @@ -149,8 +149,8 @@ ], containing_type=None, serialized_options=None, - serialized_start=3400, - serialized_end=3441, + serialized_start=3878, + serialized_end=3919, ) _sym_db.RegisterEnumDescriptor(_FUNCTIONSTATE) @@ -192,8 +192,8 @@ ], containing_type=None, serialized_options=None, - serialized_start=785, - serialized_end=824, + serialized_start=844, + serialized_end=883, ) _sym_db.RegisterEnumDescriptor(_FUNCTIONDETAILS_RUNTIME) @@ -227,11 +227,46 @@ ], containing_type=None, serialized_options=None, - serialized_start=826, - serialized_end=890, + serialized_start=885, + serialized_end=949, ) _sym_db.RegisterEnumDescriptor(_FUNCTIONDETAILS_COMPONENTTYPE) +_CRYPTOSPEC_FAILUREACTION = _descriptor.EnumDescriptor( + name='FailureAction', + full_name='proto.CryptoSpec.FailureAction', + filename=None, + file=DESCRIPTOR, + create_key=_descriptor._internal_create_key, + values=[ + _descriptor.EnumValueDescriptor( + name='FAIL', index=0, number=0, + serialized_options=None, + type=None, + create_key=_descriptor._internal_create_key), + _descriptor.EnumValueDescriptor( + name='DISCARD', index=1, number=1, + serialized_options=None, + type=None, + create_key=_descriptor._internal_create_key), + _descriptor.EnumValueDescriptor( + name='CONSUME', index=2, number=2, + serialized_options=None, + type=None, + create_key=_descriptor._internal_create_key), + _descriptor.EnumValueDescriptor( + name='SEND', index=3, number=10, + serialized_options=None, + type=None, + create_key=_descriptor._internal_create_key), + ], + containing_type=None, + serialized_options=None, + serialized_start=1873, + serialized_end=1934, +) +_sym_db.RegisterEnumDescriptor(_CRYPTOSPEC_FAILUREACTION) + _RESOURCES = _descriptor.Descriptor( name='Resources', @@ -480,6 +515,13 @@ message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), + _descriptor.FieldDescriptor( + name='subscriptionPosition', full_name='proto.FunctionDetails.subscriptionPosition', index=22, + number=23, type=14, cpp_type=8, label=1, + has_default_value=False, default_value=0, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), ], extensions=[ ], @@ -495,7 +537,7 @@ oneofs=[ ], serialized_start=147, - serialized_end=890, + serialized_end=949, ) @@ -526,8 +568,8 @@ extension_ranges=[], oneofs=[ ], - serialized_start=1185, - serialized_end=1219, + serialized_start=1283, + serialized_end=1317, ) _CONSUMERSPEC_SCHEMAPROPERTIESENTRY = _descriptor.Descriptor( @@ -564,8 +606,8 @@ extension_ranges=[], oneofs=[ ], - serialized_start=1221, - serialized_end=1276, + serialized_start=1319, + serialized_end=1374, ) _CONSUMERSPEC_CONSUMERPROPERTIESENTRY = _descriptor.Descriptor( @@ -602,8 +644,8 @@ extension_ranges=[], oneofs=[ ], - serialized_start=1278, - serialized_end=1335, + serialized_start=1376, + serialized_end=1433, ) _CONSUMERSPEC = _descriptor.Descriptor( @@ -656,6 +698,13 @@ message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), + _descriptor.FieldDescriptor( + name='cryptoSpec', full_name='proto.ConsumerSpec.cryptoSpec', index=6, + number=7, type=11, cpp_type=10, label=1, + has_default_value=False, default_value=None, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), ], extensions=[ ], @@ -668,8 +717,8 @@ extension_ranges=[], oneofs=[ ], - serialized_start=893, - serialized_end=1335, + serialized_start=952, + serialized_end=1433, ) @@ -702,6 +751,20 @@ message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), + _descriptor.FieldDescriptor( + name='cryptoSpec', full_name='proto.ProducerSpec.cryptoSpec', index=3, + number=4, type=11, cpp_type=10, label=1, + has_default_value=False, default_value=None, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), + _descriptor.FieldDescriptor( + name='batchBuilder', full_name='proto.ProducerSpec.batchBuilder', index=4, + number=5, type=9, cpp_type=9, label=1, + has_default_value=False, default_value=b"".decode('utf-8'), + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), ], extensions=[ ], @@ -714,8 +777,69 @@ extension_ranges=[], oneofs=[ ], - serialized_start=1337, - serialized_end=1456, + serialized_start=1436, + serialized_end=1616, +) + + +_CRYPTOSPEC = _descriptor.Descriptor( + name='CryptoSpec', + full_name='proto.CryptoSpec', + filename=None, + file=DESCRIPTOR, + containing_type=None, + create_key=_descriptor._internal_create_key, + fields=[ + _descriptor.FieldDescriptor( + name='cryptoKeyReaderClassName', full_name='proto.CryptoSpec.cryptoKeyReaderClassName', index=0, + number=1, type=9, cpp_type=9, label=1, + has_default_value=False, default_value=b"".decode('utf-8'), + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), + _descriptor.FieldDescriptor( + name='cryptoKeyReaderConfig', full_name='proto.CryptoSpec.cryptoKeyReaderConfig', index=1, + number=2, type=9, cpp_type=9, label=1, + has_default_value=False, default_value=b"".decode('utf-8'), + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), + _descriptor.FieldDescriptor( + name='producerEncryptionKeyName', full_name='proto.CryptoSpec.producerEncryptionKeyName', index=2, + number=3, type=9, cpp_type=9, label=3, + has_default_value=False, default_value=[], + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), + _descriptor.FieldDescriptor( + name='producerCryptoFailureAction', full_name='proto.CryptoSpec.producerCryptoFailureAction', index=3, + number=4, type=14, cpp_type=8, label=1, + has_default_value=False, default_value=0, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), + _descriptor.FieldDescriptor( + name='consumerCryptoFailureAction', full_name='proto.CryptoSpec.consumerCryptoFailureAction', index=4, + number=5, type=14, cpp_type=8, label=1, + has_default_value=False, default_value=0, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), + ], + extensions=[ + ], + nested_types=[], + enum_types=[ + _CRYPTOSPEC_FAILUREACTION, + ], + serialized_options=None, + is_extendable=False, + syntax='proto3', + extension_ranges=[], + oneofs=[ + ], + serialized_start=1619, + serialized_end=1934, ) @@ -753,8 +877,8 @@ extension_ranges=[], oneofs=[ ], - serialized_start=1936, - serialized_end=1997, + serialized_start=2414, + serialized_end=2475, ) _SOURCESPEC_INPUTSPECSENTRY = _descriptor.Descriptor( @@ -791,8 +915,8 @@ extension_ranges=[], oneofs=[ ], - serialized_start=1999, - serialized_end=2069, + serialized_start=2477, + serialized_end=2547, ) _SOURCESPEC = _descriptor.Descriptor( @@ -906,8 +1030,8 @@ extension_ranges=[], oneofs=[ ], - serialized_start=1459, - serialized_end=2069, + serialized_start=1937, + serialized_end=2547, ) @@ -945,8 +1069,8 @@ extension_ranges=[], oneofs=[ ], - serialized_start=1221, - serialized_end=1276, + serialized_start=1319, + serialized_end=1374, ) _SINKSPEC_CONSUMERPROPERTIESENTRY = _descriptor.Descriptor( @@ -983,8 +1107,8 @@ extension_ranges=[], oneofs=[ ], - serialized_start=1278, - serialized_end=1335, + serialized_start=1376, + serialized_end=1433, ) _SINKSPEC = _descriptor.Descriptor( @@ -1084,8 +1208,8 @@ extension_ranges=[], oneofs=[ ], - serialized_start=2072, - serialized_end=2548, + serialized_start=2550, + serialized_end=3026, ) @@ -1123,8 +1247,8 @@ extension_ranges=[], oneofs=[ ], - serialized_start=2550, - serialized_end=2622, + serialized_start=3028, + serialized_end=3100, ) @@ -1162,8 +1286,8 @@ extension_ranges=[], oneofs=[ ], - serialized_start=2918, - serialized_end=2993, + serialized_start=3396, + serialized_end=3471, ) _FUNCTIONMETADATA = _descriptor.Descriptor( @@ -1228,8 +1352,8 @@ extension_ranges=[], oneofs=[ ], - serialized_start=2625, - serialized_end=2993, + serialized_start=3103, + serialized_end=3471, ) @@ -1267,8 +1391,8 @@ extension_ranges=[], oneofs=[ ], - serialized_start=2995, - serialized_end=3055, + serialized_start=3473, + serialized_end=3533, ) @@ -1306,8 +1430,8 @@ extension_ranges=[], oneofs=[ ], - serialized_start=3057, - serialized_end=3138, + serialized_start=3535, + serialized_end=3616, ) @@ -1345,8 +1469,8 @@ extension_ranges=[], oneofs=[ ], - serialized_start=3140, - serialized_end=3205, + serialized_start=3618, + serialized_end=3683, ) _FUNCTIONDETAILS.fields_by_name['processingGuarantees'].enum_type = _PROCESSINGGUARANTEES @@ -1356,6 +1480,7 @@ _FUNCTIONDETAILS.fields_by_name['resources'].message_type = _RESOURCES _FUNCTIONDETAILS.fields_by_name['retryDetails'].message_type = _RETRYDETAILS _FUNCTIONDETAILS.fields_by_name['componentType'].enum_type = _FUNCTIONDETAILS_COMPONENTTYPE +_FUNCTIONDETAILS.fields_by_name['subscriptionPosition'].enum_type = _SUBSCRIPTIONPOSITION _FUNCTIONDETAILS_RUNTIME.containing_type = _FUNCTIONDETAILS _FUNCTIONDETAILS_COMPONENTTYPE.containing_type = _FUNCTIONDETAILS _CONSUMERSPEC_RECEIVERQUEUESIZE.containing_type = _CONSUMERSPEC @@ -1364,6 +1489,11 @@ _CONSUMERSPEC.fields_by_name['receiverQueueSize'].message_type = _CONSUMERSPEC_RECEIVERQUEUESIZE _CONSUMERSPEC.fields_by_name['schemaProperties'].message_type = _CONSUMERSPEC_SCHEMAPROPERTIESENTRY _CONSUMERSPEC.fields_by_name['consumerProperties'].message_type = _CONSUMERSPEC_CONSUMERPROPERTIESENTRY +_CONSUMERSPEC.fields_by_name['cryptoSpec'].message_type = _CRYPTOSPEC +_PRODUCERSPEC.fields_by_name['cryptoSpec'].message_type = _CRYPTOSPEC +_CRYPTOSPEC.fields_by_name['producerCryptoFailureAction'].enum_type = _CRYPTOSPEC_FAILUREACTION +_CRYPTOSPEC.fields_by_name['consumerCryptoFailureAction'].enum_type = _CRYPTOSPEC_FAILUREACTION +_CRYPTOSPEC_FAILUREACTION.containing_type = _CRYPTOSPEC _SOURCESPEC_TOPICSTOSERDECLASSNAMEENTRY.containing_type = _SOURCESPEC _SOURCESPEC_INPUTSPECSENTRY.fields_by_name['value'].message_type = _CONSUMERSPEC _SOURCESPEC_INPUTSPECSENTRY.containing_type = _SOURCESPEC @@ -1389,6 +1519,7 @@ DESCRIPTOR.message_types_by_name['FunctionDetails'] = _FUNCTIONDETAILS DESCRIPTOR.message_types_by_name['ConsumerSpec'] = _CONSUMERSPEC DESCRIPTOR.message_types_by_name['ProducerSpec'] = _PRODUCERSPEC +DESCRIPTOR.message_types_by_name['CryptoSpec'] = _CRYPTOSPEC DESCRIPTOR.message_types_by_name['SourceSpec'] = _SOURCESPEC DESCRIPTOR.message_types_by_name['SinkSpec'] = _SINKSPEC DESCRIPTOR.message_types_by_name['PackageLocationMetaData'] = _PACKAGELOCATIONMETADATA @@ -1461,6 +1592,13 @@ }) _sym_db.RegisterMessage(ProducerSpec) +CryptoSpec = _reflection.GeneratedProtocolMessageType('CryptoSpec', (_message.Message,), { + 'DESCRIPTOR' : _CRYPTOSPEC, + '__module__' : 'Function_pb2' + # @@protoc_insertion_point(class_scope:proto.CryptoSpec) + }) +_sym_db.RegisterMessage(CryptoSpec) + SourceSpec = _reflection.GeneratedProtocolMessageType('SourceSpec', (_message.Message,), { 'TopicsToSerDeClassNameEntry' : _reflection.GeneratedProtocolMessageType('TopicsToSerDeClassNameEntry', (_message.Message,), { diff --git a/pulsar-functions/instance/src/main/python/python_instance.py b/pulsar-functions/instance/src/main/python/python_instance.py index d6ff4fb55d08d..8c000f0c856fd 100644 --- a/pulsar-functions/instance/src/main/python/python_instance.py +++ b/pulsar-functions/instance/src/main/python/python_instance.py @@ -309,10 +309,18 @@ def setup_producer(self): len(self.instance_config.function_details.sink.topic) > 0: Log.debug("Setting up producer for topic %s" % self.instance_config.function_details.sink.topic) + batch_type = BatchingType.Default + if self.instance_config.function_details.sink.producerSpec.batchBuilder != None and \ + len(self.instance_config.function_details.sink.producerSpec.batchBuilder) > 0: + batch_builder = self.instance_config.function_details.sink.producerSpec.batchBuilder + if batch_builder == "KEY_BASED": + batch_type = BatchingType.KeyBased + self.producer = self.pulsar_client.create_producer( str(self.instance_config.function_details.sink.topic), block_if_queue_full=True, batching_enabled=True, + batching_type=batch_type, batching_max_publish_delay_ms=10, compression_type=pulsar.CompressionType.LZ4, # set send timeout to be infinity to prevent potential deadlock with consumer From 07a92933272c626fdb882fb076bbd7b33d2d32c5 Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Wed, 18 Nov 2020 17:23:38 +0800 Subject: [PATCH 5/9] fix ci test error Signed-off-by: xiaolong.ran --- .../integration/functions/PulsarFunctionsTest.java | 12 +++--------- 1 file changed, 3 insertions(+), 9 deletions(-) diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java index 723cd46e96797..aaa435b082350 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java @@ -26,15 +26,7 @@ import org.apache.commons.lang3.StringUtils; import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.admin.PulsarAdminException; -import org.apache.pulsar.client.api.Consumer; -import org.apache.pulsar.client.api.Message; -import org.apache.pulsar.client.api.MessageId; -import org.apache.pulsar.client.api.Producer; -import org.apache.pulsar.client.api.PulsarClient; -import org.apache.pulsar.client.api.Reader; -import org.apache.pulsar.client.api.Schema; -import org.apache.pulsar.client.api.SubscriptionInitialPosition; -import org.apache.pulsar.client.api.SubscriptionType; +import org.apache.pulsar.client.api.*; import org.apache.pulsar.client.impl.PulsarClientImpl; import org.apache.pulsar.client.impl.schema.AvroSchema; import org.apache.pulsar.client.impl.schema.KeyValueSchema; @@ -981,6 +973,8 @@ public void testFunctionLocalRun(Runtime runtime) throws Exception { @Cleanup Producer producer = client.newProducer(Schema.BYTES) .topic(inputTopicName) + .enableBatching(true) + .batcherBuilder(BatcherBuilder.DEFAULT) .create(); for (int i = 0; i < numMessages; i++) { From d142ed012eda1a5de97bcc4118141bc7305c0e4a Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Wed, 18 Nov 2020 22:19:15 +0800 Subject: [PATCH 6/9] update proto file Signed-off-by: xiaolong.ran --- .../main/python/InstanceCommunication_pb2.py | 306 +++++++++++------- .../python/InstanceCommunication_pb2_grpc.py | 286 ++++++++++------ 2 files changed, 382 insertions(+), 210 deletions(-) diff --git a/pulsar-functions/instance/src/main/python/InstanceCommunication_pb2.py b/pulsar-functions/instance/src/main/python/InstanceCommunication_pb2.py index d665e9815ab9e..d1ba3bb0ed094 100644 --- a/pulsar-functions/instance/src/main/python/InstanceCommunication_pb2.py +++ b/pulsar-functions/instance/src/main/python/InstanceCommunication_pb2.py @@ -19,14 +19,11 @@ # Generated by the protocol buffer compiler. DO NOT EDIT! # source: InstanceCommunication.proto - -import sys -_b=sys.version_info[0]<3 and (lambda x:x) or (lambda x:x.encode('latin1')) +"""Generated protocol buffer code.""" from google.protobuf import descriptor as _descriptor from google.protobuf import message as _message from google.protobuf import reflection as _reflection from google.protobuf import symbol_database as _symbol_database -from google.protobuf import descriptor_pb2 # @@protoc_insertion_point(imports) _sym_db = _symbol_database.Default() @@ -39,7 +36,9 @@ name='InstanceCommunication.proto', package='proto', syntax='proto3', - serialized_pb=_b('\n\x1bInstanceCommunication.proto\x12\x05proto\x1a\x1bgoogle/protobuf/empty.proto\"\xf6\x03\n\x0e\x46unctionStatus\x12\x0f\n\x07running\x18\x01 \x01(\x08\x12\x18\n\x10\x66\x61ilureException\x18\x02 \x01(\t\x12\x13\n\x0bnumRestarts\x18\x03 \x01(\x03\x12\x13\n\x0bnumReceived\x18\x11 \x01(\x03\x12 \n\x18numSuccessfullyProcessed\x18\x05 \x01(\x03\x12\x19\n\x11numUserExceptions\x18\x06 \x01(\x03\x12H\n\x14latestUserExceptions\x18\x07 \x03(\x0b\x32*.proto.FunctionStatus.ExceptionInformation\x12\x1b\n\x13numSystemExceptions\x18\x08 \x01(\x03\x12J\n\x16latestSystemExceptions\x18\t \x03(\x0b\x32*.proto.FunctionStatus.ExceptionInformation\x12\x16\n\x0e\x61verageLatency\x18\x0c \x01(\x01\x12\x1a\n\x12lastInvocationTime\x18\r \x01(\x03\x12\x12\n\ninstanceId\x18\x0e \x01(\t\x12\x10\n\x08workerId\x18\x10 \x01(\t\x1a\x45\n\x14\x45xceptionInformation\x12\x17\n\x0f\x65xceptionString\x18\x01 \x01(\t\x12\x14\n\x0cmsSinceEpoch\x18\x02 \x01(\x03\"\xd0\x03\n\x0bMetricsData\x12\x15\n\rreceivedTotal\x18\x02 \x01(\x03\x12\x1a\n\x12receivedTotal_1min\x18\n \x01(\x03\x12\"\n\x1aprocessedSuccessfullyTotal\x18\x04 \x01(\x03\x12\'\n\x1fprocessedSuccessfullyTotal_1min\x18\x0c \x01(\x03\x12\x1d\n\x15systemExceptionsTotal\x18\x05 \x01(\x03\x12\"\n\x1asystemExceptionsTotal_1min\x18\r \x01(\x03\x12\x1b\n\x13userExceptionsTotal\x18\x06 \x01(\x03\x12 \n\x18userExceptionsTotal_1min\x18\x0e \x01(\x03\x12\x19\n\x11\x61vgProcessLatency\x18\x07 \x01(\x01\x12\x1e\n\x16\x61vgProcessLatency_1min\x18\x0f \x01(\x01\x12\x16\n\x0elastInvocation\x18\x08 \x01(\x03\x12\x38\n\x0buserMetrics\x18\t \x03(\x0b\x32#.proto.MetricsData.UserMetricsEntry\x1a\x32\n\x10UserMetricsEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\x01:\x02\x38\x01\"$\n\x11HealthCheckResult\x12\x0f\n\x07success\x18\x01 \x01(\x08\"\x98\x01\n\x07Metrics\x12/\n\x07metrics\x18\x01 \x03(\x0b\x32\x1e.proto.Metrics.InstanceMetrics\x1a\\\n\x0fInstanceMetrics\x12\x0c\n\x04name\x18\x01 \x01(\t\x12\x12\n\ninstanceId\x18\x02 \x01(\x05\x12\'\n\x0bmetricsData\x18\x03 \x01(\x0b\x32\x12.proto.MetricsData2\xdc\x02\n\x0fInstanceControl\x12\x44\n\x11GetFunctionStatus\x12\x16.google.protobuf.Empty\x1a\x15.proto.FunctionStatus\"\x00\x12\x42\n\x12GetAndResetMetrics\x12\x16.google.protobuf.Empty\x1a\x12.proto.MetricsData\"\x00\x12@\n\x0cResetMetrics\x12\x16.google.protobuf.Empty\x1a\x16.google.protobuf.Empty\"\x00\x12:\n\nGetMetrics\x12\x16.google.protobuf.Empty\x1a\x12.proto.MetricsData\"\x00\x12\x41\n\x0bHealthCheck\x12\x16.google.protobuf.Empty\x1a\x18.proto.HealthCheckResult\"\x00\x42:\n!org.apache.pulsar.functions.protoB\x15InstanceCommunicationb\x06proto3') + serialized_options=b'\n!org.apache.pulsar.functions.protoB\025InstanceCommunication', + create_key=_descriptor._internal_create_key, + serialized_pb=b'\n\x1bInstanceCommunication.proto\x12\x05proto\x1a\x1bgoogle/protobuf/empty.proto\"\xc4\x05\n\x0e\x46unctionStatus\x12\x0f\n\x07running\x18\x01 \x01(\x08\x12\x18\n\x10\x66\x61ilureException\x18\x02 \x01(\t\x12\x13\n\x0bnumRestarts\x18\x03 \x01(\x03\x12\x13\n\x0bnumReceived\x18\x11 \x01(\x03\x12 \n\x18numSuccessfullyProcessed\x18\x05 \x01(\x03\x12\x19\n\x11numUserExceptions\x18\x06 \x01(\x03\x12H\n\x14latestUserExceptions\x18\x07 \x03(\x0b\x32*.proto.FunctionStatus.ExceptionInformation\x12\x1b\n\x13numSystemExceptions\x18\x08 \x01(\x03\x12J\n\x16latestSystemExceptions\x18\t \x03(\x0b\x32*.proto.FunctionStatus.ExceptionInformation\x12\x1b\n\x13numSourceExceptions\x18\x12 \x01(\x03\x12J\n\x16latestSourceExceptions\x18\x13 \x03(\x0b\x32*.proto.FunctionStatus.ExceptionInformation\x12\x19\n\x11numSinkExceptions\x18\x14 \x01(\x03\x12H\n\x14latestSinkExceptions\x18\x15 \x03(\x0b\x32*.proto.FunctionStatus.ExceptionInformation\x12\x16\n\x0e\x61verageLatency\x18\x0c \x01(\x01\x12\x1a\n\x12lastInvocationTime\x18\r \x01(\x03\x12\x12\n\ninstanceId\x18\x0e \x01(\t\x12\x10\n\x08workerId\x18\x10 \x01(\t\x1a\x45\n\x14\x45xceptionInformation\x12\x17\n\x0f\x65xceptionString\x18\x01 \x01(\t\x12\x14\n\x0cmsSinceEpoch\x18\x02 \x01(\x03\"V\n\x12\x46unctionStatusList\x12\r\n\x05\x65rror\x18\x02 \x01(\t\x12\x31\n\x12\x66unctionStatusList\x18\x01 \x03(\x0b\x32\x15.proto.FunctionStatus\"\xd0\x03\n\x0bMetricsData\x12\x15\n\rreceivedTotal\x18\x02 \x01(\x03\x12\x1a\n\x12receivedTotal_1min\x18\n \x01(\x03\x12\"\n\x1aprocessedSuccessfullyTotal\x18\x04 \x01(\x03\x12\'\n\x1fprocessedSuccessfullyTotal_1min\x18\x0c \x01(\x03\x12\x1d\n\x15systemExceptionsTotal\x18\x05 \x01(\x03\x12\"\n\x1asystemExceptionsTotal_1min\x18\r \x01(\x03\x12\x1b\n\x13userExceptionsTotal\x18\x06 \x01(\x03\x12 \n\x18userExceptionsTotal_1min\x18\x0e \x01(\x03\x12\x19\n\x11\x61vgProcessLatency\x18\x07 \x01(\x01\x12\x1e\n\x16\x61vgProcessLatency_1min\x18\x0f \x01(\x01\x12\x16\n\x0elastInvocation\x18\x08 \x01(\x03\x12\x38\n\x0buserMetrics\x18\t \x03(\x0b\x32#.proto.MetricsData.UserMetricsEntry\x1a\x32\n\x10UserMetricsEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\x01:\x02\x38\x01\"$\n\x11HealthCheckResult\x12\x0f\n\x07success\x18\x01 \x01(\x08\"\x98\x01\n\x07Metrics\x12/\n\x07metrics\x18\x01 \x03(\x0b\x32\x1e.proto.Metrics.InstanceMetrics\x1a\\\n\x0fInstanceMetrics\x12\x0c\n\x04name\x18\x01 \x01(\t\x12\x12\n\ninstanceId\x18\x02 \x01(\x05\x12\'\n\x0bmetricsData\x18\x03 \x01(\x0b\x32\x12.proto.MetricsData2\xdc\x02\n\x0fInstanceControl\x12\x44\n\x11GetFunctionStatus\x12\x16.google.protobuf.Empty\x1a\x15.proto.FunctionStatus\"\x00\x12\x42\n\x12GetAndResetMetrics\x12\x16.google.protobuf.Empty\x1a\x12.proto.MetricsData\"\x00\x12@\n\x0cResetMetrics\x12\x16.google.protobuf.Empty\x1a\x16.google.protobuf.Empty\"\x00\x12:\n\nGetMetrics\x12\x16.google.protobuf.Empty\x1a\x12.proto.MetricsData\"\x00\x12\x41\n\x0bHealthCheck\x12\x16.google.protobuf.Empty\x1a\x18.proto.HealthCheckResult\"\x00\x42:\n!org.apache.pulsar.functions.protoB\x15InstanceCommunicationb\x06proto3' , dependencies=[google_dot_protobuf_dot_empty__pb2.DESCRIPTOR,]) @@ -52,35 +51,36 @@ filename=None, file=DESCRIPTOR, containing_type=None, + create_key=_descriptor._internal_create_key, fields=[ _descriptor.FieldDescriptor( name='exceptionString', full_name='proto.FunctionStatus.ExceptionInformation.exceptionString', index=0, number=1, type=9, cpp_type=9, label=1, - has_default_value=False, default_value=_b("").decode('utf-8'), + has_default_value=False, default_value=b"".decode('utf-8'), message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), _descriptor.FieldDescriptor( name='msSinceEpoch', full_name='proto.FunctionStatus.ExceptionInformation.msSinceEpoch', index=1, number=2, type=3, cpp_type=2, label=1, has_default_value=False, default_value=0, message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), ], extensions=[ ], nested_types=[], enum_types=[ ], - options=None, + serialized_options=None, is_extendable=False, syntax='proto3', extension_ranges=[], oneofs=[ ], - serialized_start=501, - serialized_end=570, + serialized_start=707, + serialized_end=776, ) _FUNCTIONSTATUS = _descriptor.Descriptor( @@ -89,6 +89,7 @@ filename=None, file=DESCRIPTOR, containing_type=None, + create_key=_descriptor._internal_create_key, fields=[ _descriptor.FieldDescriptor( name='running', full_name='proto.FunctionStatus.running', index=0, @@ -96,105 +97,172 @@ has_default_value=False, default_value=False, message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), _descriptor.FieldDescriptor( name='failureException', full_name='proto.FunctionStatus.failureException', index=1, number=2, type=9, cpp_type=9, label=1, - has_default_value=False, default_value=_b("").decode('utf-8'), + has_default_value=False, default_value=b"".decode('utf-8'), message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), _descriptor.FieldDescriptor( name='numRestarts', full_name='proto.FunctionStatus.numRestarts', index=2, number=3, type=3, cpp_type=2, label=1, has_default_value=False, default_value=0, message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), _descriptor.FieldDescriptor( name='numReceived', full_name='proto.FunctionStatus.numReceived', index=3, number=17, type=3, cpp_type=2, label=1, has_default_value=False, default_value=0, message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), _descriptor.FieldDescriptor( name='numSuccessfullyProcessed', full_name='proto.FunctionStatus.numSuccessfullyProcessed', index=4, number=5, type=3, cpp_type=2, label=1, has_default_value=False, default_value=0, message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), _descriptor.FieldDescriptor( name='numUserExceptions', full_name='proto.FunctionStatus.numUserExceptions', index=5, number=6, type=3, cpp_type=2, label=1, has_default_value=False, default_value=0, message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), _descriptor.FieldDescriptor( name='latestUserExceptions', full_name='proto.FunctionStatus.latestUserExceptions', index=6, number=7, type=11, cpp_type=10, label=3, has_default_value=False, default_value=[], message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), _descriptor.FieldDescriptor( name='numSystemExceptions', full_name='proto.FunctionStatus.numSystemExceptions', index=7, number=8, type=3, cpp_type=2, label=1, has_default_value=False, default_value=0, message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), _descriptor.FieldDescriptor( name='latestSystemExceptions', full_name='proto.FunctionStatus.latestSystemExceptions', index=8, number=9, type=11, cpp_type=10, label=3, has_default_value=False, default_value=[], message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), + _descriptor.FieldDescriptor( + name='numSourceExceptions', full_name='proto.FunctionStatus.numSourceExceptions', index=9, + number=18, type=3, cpp_type=2, label=1, + has_default_value=False, default_value=0, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), + _descriptor.FieldDescriptor( + name='latestSourceExceptions', full_name='proto.FunctionStatus.latestSourceExceptions', index=10, + number=19, type=11, cpp_type=10, label=3, + has_default_value=False, default_value=[], + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), _descriptor.FieldDescriptor( - name='averageLatency', full_name='proto.FunctionStatus.averageLatency', index=9, + name='numSinkExceptions', full_name='proto.FunctionStatus.numSinkExceptions', index=11, + number=20, type=3, cpp_type=2, label=1, + has_default_value=False, default_value=0, + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), + _descriptor.FieldDescriptor( + name='latestSinkExceptions', full_name='proto.FunctionStatus.latestSinkExceptions', index=12, + number=21, type=11, cpp_type=10, label=3, + has_default_value=False, default_value=[], + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), + _descriptor.FieldDescriptor( + name='averageLatency', full_name='proto.FunctionStatus.averageLatency', index=13, number=12, type=1, cpp_type=5, label=1, has_default_value=False, default_value=float(0), message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), _descriptor.FieldDescriptor( - name='lastInvocationTime', full_name='proto.FunctionStatus.lastInvocationTime', index=10, + name='lastInvocationTime', full_name='proto.FunctionStatus.lastInvocationTime', index=14, number=13, type=3, cpp_type=2, label=1, has_default_value=False, default_value=0, message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), _descriptor.FieldDescriptor( - name='instanceId', full_name='proto.FunctionStatus.instanceId', index=11, + name='instanceId', full_name='proto.FunctionStatus.instanceId', index=15, number=14, type=9, cpp_type=9, label=1, - has_default_value=False, default_value=_b("").decode('utf-8'), + has_default_value=False, default_value=b"".decode('utf-8'), message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), _descriptor.FieldDescriptor( - name='workerId', full_name='proto.FunctionStatus.workerId', index=12, + name='workerId', full_name='proto.FunctionStatus.workerId', index=16, number=16, type=9, cpp_type=9, label=1, - has_default_value=False, default_value=_b("").decode('utf-8'), + has_default_value=False, default_value=b"".decode('utf-8'), message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), ], extensions=[ ], nested_types=[_FUNCTIONSTATUS_EXCEPTIONINFORMATION, ], enum_types=[ ], - options=None, + serialized_options=None, is_extendable=False, syntax='proto3', extension_ranges=[], oneofs=[ ], serialized_start=68, - serialized_end=570, + serialized_end=776, +) + + +_FUNCTIONSTATUSLIST = _descriptor.Descriptor( + name='FunctionStatusList', + full_name='proto.FunctionStatusList', + filename=None, + file=DESCRIPTOR, + containing_type=None, + create_key=_descriptor._internal_create_key, + fields=[ + _descriptor.FieldDescriptor( + name='error', full_name='proto.FunctionStatusList.error', index=0, + number=2, type=9, cpp_type=9, label=1, + has_default_value=False, default_value=b"".decode('utf-8'), + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), + _descriptor.FieldDescriptor( + name='functionStatusList', full_name='proto.FunctionStatusList.functionStatusList', index=1, + number=1, type=11, cpp_type=10, label=3, + has_default_value=False, default_value=[], + message_type=None, enum_type=None, containing_type=None, + is_extension=False, extension_scope=None, + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), + ], + extensions=[ + ], + nested_types=[], + enum_types=[ + ], + serialized_options=None, + is_extendable=False, + syntax='proto3', + extension_ranges=[], + oneofs=[ + ], + serialized_start=778, + serialized_end=864, ) @@ -204,35 +272,36 @@ filename=None, file=DESCRIPTOR, containing_type=None, + create_key=_descriptor._internal_create_key, fields=[ _descriptor.FieldDescriptor( name='key', full_name='proto.MetricsData.UserMetricsEntry.key', index=0, number=1, type=9, cpp_type=9, label=1, - has_default_value=False, default_value=_b("").decode('utf-8'), + has_default_value=False, default_value=b"".decode('utf-8'), message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), _descriptor.FieldDescriptor( name='value', full_name='proto.MetricsData.UserMetricsEntry.value', index=1, number=2, type=1, cpp_type=5, label=1, has_default_value=False, default_value=float(0), message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), ], extensions=[ ], nested_types=[], enum_types=[ ], - options=_descriptor._ParseOptions(descriptor_pb2.MessageOptions(), _b('8\001')), + serialized_options=b'8\001', is_extendable=False, syntax='proto3', extension_ranges=[], oneofs=[ ], - serialized_start=987, - serialized_end=1037, + serialized_start=1281, + serialized_end=1331, ) _METRICSDATA = _descriptor.Descriptor( @@ -241,6 +310,7 @@ filename=None, file=DESCRIPTOR, containing_type=None, + create_key=_descriptor._internal_create_key, fields=[ _descriptor.FieldDescriptor( name='receivedTotal', full_name='proto.MetricsData.receivedTotal', index=0, @@ -248,98 +318,98 @@ has_default_value=False, default_value=0, message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), _descriptor.FieldDescriptor( name='receivedTotal_1min', full_name='proto.MetricsData.receivedTotal_1min', index=1, number=10, type=3, cpp_type=2, label=1, has_default_value=False, default_value=0, message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), _descriptor.FieldDescriptor( name='processedSuccessfullyTotal', full_name='proto.MetricsData.processedSuccessfullyTotal', index=2, number=4, type=3, cpp_type=2, label=1, has_default_value=False, default_value=0, message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), _descriptor.FieldDescriptor( name='processedSuccessfullyTotal_1min', full_name='proto.MetricsData.processedSuccessfullyTotal_1min', index=3, number=12, type=3, cpp_type=2, label=1, has_default_value=False, default_value=0, message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), _descriptor.FieldDescriptor( name='systemExceptionsTotal', full_name='proto.MetricsData.systemExceptionsTotal', index=4, number=5, type=3, cpp_type=2, label=1, has_default_value=False, default_value=0, message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), _descriptor.FieldDescriptor( name='systemExceptionsTotal_1min', full_name='proto.MetricsData.systemExceptionsTotal_1min', index=5, number=13, type=3, cpp_type=2, label=1, has_default_value=False, default_value=0, message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), _descriptor.FieldDescriptor( name='userExceptionsTotal', full_name='proto.MetricsData.userExceptionsTotal', index=6, number=6, type=3, cpp_type=2, label=1, has_default_value=False, default_value=0, message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), _descriptor.FieldDescriptor( name='userExceptionsTotal_1min', full_name='proto.MetricsData.userExceptionsTotal_1min', index=7, number=14, type=3, cpp_type=2, label=1, has_default_value=False, default_value=0, message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), _descriptor.FieldDescriptor( name='avgProcessLatency', full_name='proto.MetricsData.avgProcessLatency', index=8, number=7, type=1, cpp_type=5, label=1, has_default_value=False, default_value=float(0), message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), _descriptor.FieldDescriptor( name='avgProcessLatency_1min', full_name='proto.MetricsData.avgProcessLatency_1min', index=9, number=15, type=1, cpp_type=5, label=1, has_default_value=False, default_value=float(0), message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), _descriptor.FieldDescriptor( name='lastInvocation', full_name='proto.MetricsData.lastInvocation', index=10, number=8, type=3, cpp_type=2, label=1, has_default_value=False, default_value=0, message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), _descriptor.FieldDescriptor( name='userMetrics', full_name='proto.MetricsData.userMetrics', index=11, number=9, type=11, cpp_type=10, label=3, has_default_value=False, default_value=[], message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), ], extensions=[ ], nested_types=[_METRICSDATA_USERMETRICSENTRY, ], enum_types=[ ], - options=None, + serialized_options=None, is_extendable=False, syntax='proto3', extension_ranges=[], oneofs=[ ], - serialized_start=573, - serialized_end=1037, + serialized_start=867, + serialized_end=1331, ) @@ -349,6 +419,7 @@ filename=None, file=DESCRIPTOR, containing_type=None, + create_key=_descriptor._internal_create_key, fields=[ _descriptor.FieldDescriptor( name='success', full_name='proto.HealthCheckResult.success', index=0, @@ -356,21 +427,21 @@ has_default_value=False, default_value=False, message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), ], extensions=[ ], nested_types=[], enum_types=[ ], - options=None, + serialized_options=None, is_extendable=False, syntax='proto3', extension_ranges=[], oneofs=[ ], - serialized_start=1039, - serialized_end=1075, + serialized_start=1333, + serialized_end=1369, ) @@ -380,42 +451,43 @@ filename=None, file=DESCRIPTOR, containing_type=None, + create_key=_descriptor._internal_create_key, fields=[ _descriptor.FieldDescriptor( name='name', full_name='proto.Metrics.InstanceMetrics.name', index=0, number=1, type=9, cpp_type=9, label=1, - has_default_value=False, default_value=_b("").decode('utf-8'), + has_default_value=False, default_value=b"".decode('utf-8'), message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), _descriptor.FieldDescriptor( name='instanceId', full_name='proto.Metrics.InstanceMetrics.instanceId', index=1, number=2, type=5, cpp_type=1, label=1, has_default_value=False, default_value=0, message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), _descriptor.FieldDescriptor( name='metricsData', full_name='proto.Metrics.InstanceMetrics.metricsData', index=2, number=3, type=11, cpp_type=10, label=1, has_default_value=False, default_value=None, message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), ], extensions=[ ], nested_types=[], enum_types=[ ], - options=None, + serialized_options=None, is_extendable=False, syntax='proto3', extension_ranges=[], oneofs=[ ], - serialized_start=1138, - serialized_end=1230, + serialized_start=1432, + serialized_end=1524, ) _METRICS = _descriptor.Descriptor( @@ -424,6 +496,7 @@ filename=None, file=DESCRIPTOR, containing_type=None, + create_key=_descriptor._internal_create_key, fields=[ _descriptor.FieldDescriptor( name='metrics', full_name='proto.Metrics.metrics', index=0, @@ -431,103 +504,113 @@ has_default_value=False, default_value=[], message_type=None, enum_type=None, containing_type=None, is_extension=False, extension_scope=None, - options=None, file=DESCRIPTOR), + serialized_options=None, file=DESCRIPTOR, create_key=_descriptor._internal_create_key), ], extensions=[ ], nested_types=[_METRICS_INSTANCEMETRICS, ], enum_types=[ ], - options=None, + serialized_options=None, is_extendable=False, syntax='proto3', extension_ranges=[], oneofs=[ ], - serialized_start=1078, - serialized_end=1230, + serialized_start=1372, + serialized_end=1524, ) _FUNCTIONSTATUS_EXCEPTIONINFORMATION.containing_type = _FUNCTIONSTATUS _FUNCTIONSTATUS.fields_by_name['latestUserExceptions'].message_type = _FUNCTIONSTATUS_EXCEPTIONINFORMATION _FUNCTIONSTATUS.fields_by_name['latestSystemExceptions'].message_type = _FUNCTIONSTATUS_EXCEPTIONINFORMATION +_FUNCTIONSTATUS.fields_by_name['latestSourceExceptions'].message_type = _FUNCTIONSTATUS_EXCEPTIONINFORMATION +_FUNCTIONSTATUS.fields_by_name['latestSinkExceptions'].message_type = _FUNCTIONSTATUS_EXCEPTIONINFORMATION +_FUNCTIONSTATUSLIST.fields_by_name['functionStatusList'].message_type = _FUNCTIONSTATUS _METRICSDATA_USERMETRICSENTRY.containing_type = _METRICSDATA _METRICSDATA.fields_by_name['userMetrics'].message_type = _METRICSDATA_USERMETRICSENTRY _METRICS_INSTANCEMETRICS.fields_by_name['metricsData'].message_type = _METRICSDATA _METRICS_INSTANCEMETRICS.containing_type = _METRICS _METRICS.fields_by_name['metrics'].message_type = _METRICS_INSTANCEMETRICS DESCRIPTOR.message_types_by_name['FunctionStatus'] = _FUNCTIONSTATUS +DESCRIPTOR.message_types_by_name['FunctionStatusList'] = _FUNCTIONSTATUSLIST DESCRIPTOR.message_types_by_name['MetricsData'] = _METRICSDATA DESCRIPTOR.message_types_by_name['HealthCheckResult'] = _HEALTHCHECKRESULT DESCRIPTOR.message_types_by_name['Metrics'] = _METRICS _sym_db.RegisterFileDescriptor(DESCRIPTOR) -FunctionStatus = _reflection.GeneratedProtocolMessageType('FunctionStatus', (_message.Message,), dict( +FunctionStatus = _reflection.GeneratedProtocolMessageType('FunctionStatus', (_message.Message,), { - ExceptionInformation = _reflection.GeneratedProtocolMessageType('ExceptionInformation', (_message.Message,), dict( - DESCRIPTOR = _FUNCTIONSTATUS_EXCEPTIONINFORMATION, - __module__ = 'InstanceCommunication_pb2' + 'ExceptionInformation' : _reflection.GeneratedProtocolMessageType('ExceptionInformation', (_message.Message,), { + 'DESCRIPTOR' : _FUNCTIONSTATUS_EXCEPTIONINFORMATION, + '__module__' : 'InstanceCommunication_pb2' # @@protoc_insertion_point(class_scope:proto.FunctionStatus.ExceptionInformation) - )) + }) , - DESCRIPTOR = _FUNCTIONSTATUS, - __module__ = 'InstanceCommunication_pb2' + 'DESCRIPTOR' : _FUNCTIONSTATUS, + '__module__' : 'InstanceCommunication_pb2' # @@protoc_insertion_point(class_scope:proto.FunctionStatus) - )) + }) _sym_db.RegisterMessage(FunctionStatus) _sym_db.RegisterMessage(FunctionStatus.ExceptionInformation) -MetricsData = _reflection.GeneratedProtocolMessageType('MetricsData', (_message.Message,), dict( +FunctionStatusList = _reflection.GeneratedProtocolMessageType('FunctionStatusList', (_message.Message,), { + 'DESCRIPTOR' : _FUNCTIONSTATUSLIST, + '__module__' : 'InstanceCommunication_pb2' + # @@protoc_insertion_point(class_scope:proto.FunctionStatusList) + }) +_sym_db.RegisterMessage(FunctionStatusList) + +MetricsData = _reflection.GeneratedProtocolMessageType('MetricsData', (_message.Message,), { - UserMetricsEntry = _reflection.GeneratedProtocolMessageType('UserMetricsEntry', (_message.Message,), dict( - DESCRIPTOR = _METRICSDATA_USERMETRICSENTRY, - __module__ = 'InstanceCommunication_pb2' + 'UserMetricsEntry' : _reflection.GeneratedProtocolMessageType('UserMetricsEntry', (_message.Message,), { + 'DESCRIPTOR' : _METRICSDATA_USERMETRICSENTRY, + '__module__' : 'InstanceCommunication_pb2' # @@protoc_insertion_point(class_scope:proto.MetricsData.UserMetricsEntry) - )) + }) , - DESCRIPTOR = _METRICSDATA, - __module__ = 'InstanceCommunication_pb2' + 'DESCRIPTOR' : _METRICSDATA, + '__module__' : 'InstanceCommunication_pb2' # @@protoc_insertion_point(class_scope:proto.MetricsData) - )) + }) _sym_db.RegisterMessage(MetricsData) _sym_db.RegisterMessage(MetricsData.UserMetricsEntry) -HealthCheckResult = _reflection.GeneratedProtocolMessageType('HealthCheckResult', (_message.Message,), dict( - DESCRIPTOR = _HEALTHCHECKRESULT, - __module__ = 'InstanceCommunication_pb2' +HealthCheckResult = _reflection.GeneratedProtocolMessageType('HealthCheckResult', (_message.Message,), { + 'DESCRIPTOR' : _HEALTHCHECKRESULT, + '__module__' : 'InstanceCommunication_pb2' # @@protoc_insertion_point(class_scope:proto.HealthCheckResult) - )) + }) _sym_db.RegisterMessage(HealthCheckResult) -Metrics = _reflection.GeneratedProtocolMessageType('Metrics', (_message.Message,), dict( +Metrics = _reflection.GeneratedProtocolMessageType('Metrics', (_message.Message,), { - InstanceMetrics = _reflection.GeneratedProtocolMessageType('InstanceMetrics', (_message.Message,), dict( - DESCRIPTOR = _METRICS_INSTANCEMETRICS, - __module__ = 'InstanceCommunication_pb2' + 'InstanceMetrics' : _reflection.GeneratedProtocolMessageType('InstanceMetrics', (_message.Message,), { + 'DESCRIPTOR' : _METRICS_INSTANCEMETRICS, + '__module__' : 'InstanceCommunication_pb2' # @@protoc_insertion_point(class_scope:proto.Metrics.InstanceMetrics) - )) + }) , - DESCRIPTOR = _METRICS, - __module__ = 'InstanceCommunication_pb2' + 'DESCRIPTOR' : _METRICS, + '__module__' : 'InstanceCommunication_pb2' # @@protoc_insertion_point(class_scope:proto.Metrics) - )) + }) _sym_db.RegisterMessage(Metrics) _sym_db.RegisterMessage(Metrics.InstanceMetrics) -DESCRIPTOR.has_options = True -DESCRIPTOR._options = _descriptor._ParseOptions(descriptor_pb2.FileOptions(), _b('\n!org.apache.pulsar.functions.protoB\025InstanceCommunication')) -_METRICSDATA_USERMETRICSENTRY.has_options = True -_METRICSDATA_USERMETRICSENTRY._options = _descriptor._ParseOptions(descriptor_pb2.MessageOptions(), _b('8\001')) +DESCRIPTOR._options = None +_METRICSDATA_USERMETRICSENTRY._options = None _INSTANCECONTROL = _descriptor.ServiceDescriptor( name='InstanceControl', full_name='proto.InstanceControl', file=DESCRIPTOR, index=0, - options=None, - serialized_start=1233, - serialized_end=1581, + serialized_options=None, + create_key=_descriptor._internal_create_key, + serialized_start=1527, + serialized_end=1875, methods=[ _descriptor.MethodDescriptor( name='GetFunctionStatus', @@ -536,7 +619,8 @@ containing_service=None, input_type=google_dot_protobuf_dot_empty__pb2._EMPTY, output_type=_FUNCTIONSTATUS, - options=None, + serialized_options=None, + create_key=_descriptor._internal_create_key, ), _descriptor.MethodDescriptor( name='GetAndResetMetrics', @@ -545,7 +629,8 @@ containing_service=None, input_type=google_dot_protobuf_dot_empty__pb2._EMPTY, output_type=_METRICSDATA, - options=None, + serialized_options=None, + create_key=_descriptor._internal_create_key, ), _descriptor.MethodDescriptor( name='ResetMetrics', @@ -554,7 +639,8 @@ containing_service=None, input_type=google_dot_protobuf_dot_empty__pb2._EMPTY, output_type=google_dot_protobuf_dot_empty__pb2._EMPTY, - options=None, + serialized_options=None, + create_key=_descriptor._internal_create_key, ), _descriptor.MethodDescriptor( name='GetMetrics', @@ -563,7 +649,8 @@ containing_service=None, input_type=google_dot_protobuf_dot_empty__pb2._EMPTY, output_type=_METRICSDATA, - options=None, + serialized_options=None, + create_key=_descriptor._internal_create_key, ), _descriptor.MethodDescriptor( name='HealthCheck', @@ -572,7 +659,8 @@ containing_service=None, input_type=google_dot_protobuf_dot_empty__pb2._EMPTY, output_type=_HEALTHCHECKRESULT, - options=None, + serialized_options=None, + create_key=_descriptor._internal_create_key, ), ]) _sym_db.RegisterServiceDescriptor(_INSTANCECONTROL) diff --git a/pulsar-functions/instance/src/main/python/InstanceCommunication_pb2_grpc.py b/pulsar-functions/instance/src/main/python/InstanceCommunication_pb2_grpc.py index 21730e1b52f3a..e895f53dbfa9f 100644 --- a/pulsar-functions/instance/src/main/python/InstanceCommunication_pb2_grpc.py +++ b/pulsar-functions/instance/src/main/python/InstanceCommunication_pb2_grpc.py @@ -18,6 +18,7 @@ # # Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT! +"""Client and server classes corresponding to protobuf-defined services.""" import grpc import InstanceCommunication_pb2 as InstanceCommunication__pb2 @@ -25,110 +26,193 @@ class InstanceControlStub(object): - # missing associated documentation comment in .proto file - pass - - def __init__(self, channel): - """Constructor. - - Args: - channel: A grpc.Channel. - """ - self.GetFunctionStatus = channel.unary_unary( - '/proto.InstanceControl/GetFunctionStatus', - request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, - response_deserializer=InstanceCommunication__pb2.FunctionStatus.FromString, - ) - self.GetAndResetMetrics = channel.unary_unary( - '/proto.InstanceControl/GetAndResetMetrics', - request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, - response_deserializer=InstanceCommunication__pb2.MetricsData.FromString, - ) - self.ResetMetrics = channel.unary_unary( - '/proto.InstanceControl/ResetMetrics', - request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, - response_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, - ) - self.GetMetrics = channel.unary_unary( - '/proto.InstanceControl/GetMetrics', - request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, - response_deserializer=InstanceCommunication__pb2.MetricsData.FromString, - ) - self.HealthCheck = channel.unary_unary( - '/proto.InstanceControl/HealthCheck', - request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, - response_deserializer=InstanceCommunication__pb2.HealthCheckResult.FromString, - ) + """Missing associated documentation comment in .proto file.""" + + def __init__(self, channel): + """Constructor. + + Args: + channel: A grpc.Channel. + """ + self.GetFunctionStatus = channel.unary_unary( + '/proto.InstanceControl/GetFunctionStatus', + request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, + response_deserializer=InstanceCommunication__pb2.FunctionStatus.FromString, + ) + self.GetAndResetMetrics = channel.unary_unary( + '/proto.InstanceControl/GetAndResetMetrics', + request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, + response_deserializer=InstanceCommunication__pb2.MetricsData.FromString, + ) + self.ResetMetrics = channel.unary_unary( + '/proto.InstanceControl/ResetMetrics', + request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, + response_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, + ) + self.GetMetrics = channel.unary_unary( + '/proto.InstanceControl/GetMetrics', + request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, + response_deserializer=InstanceCommunication__pb2.MetricsData.FromString, + ) + self.HealthCheck = channel.unary_unary( + '/proto.InstanceControl/HealthCheck', + request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, + response_deserializer=InstanceCommunication__pb2.HealthCheckResult.FromString, + ) class InstanceControlServicer(object): - # missing associated documentation comment in .proto file - pass - - def GetFunctionStatus(self, request, context): - # missing associated documentation comment in .proto file - pass - context.set_code(grpc.StatusCode.UNIMPLEMENTED) - context.set_details('Method not implemented!') - raise NotImplementedError('Method not implemented!') - - def GetAndResetMetrics(self, request, context): - # missing associated documentation comment in .proto file - pass - context.set_code(grpc.StatusCode.UNIMPLEMENTED) - context.set_details('Method not implemented!') - raise NotImplementedError('Method not implemented!') - - def ResetMetrics(self, request, context): - # missing associated documentation comment in .proto file - pass - context.set_code(grpc.StatusCode.UNIMPLEMENTED) - context.set_details('Method not implemented!') - raise NotImplementedError('Method not implemented!') - - def GetMetrics(self, request, context): - # missing associated documentation comment in .proto file - pass - context.set_code(grpc.StatusCode.UNIMPLEMENTED) - context.set_details('Method not implemented!') - raise NotImplementedError('Method not implemented!') - - def HealthCheck(self, request, context): - # missing associated documentation comment in .proto file - pass - context.set_code(grpc.StatusCode.UNIMPLEMENTED) - context.set_details('Method not implemented!') - raise NotImplementedError('Method not implemented!') + """Missing associated documentation comment in .proto file.""" + + def GetFunctionStatus(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + + def GetAndResetMetrics(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + + def ResetMetrics(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + + def GetMetrics(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + + def HealthCheck(self, request, context): + """Missing associated documentation comment in .proto file.""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') def add_InstanceControlServicer_to_server(servicer, server): - rpc_method_handlers = { - 'GetFunctionStatus': grpc.unary_unary_rpc_method_handler( - servicer.GetFunctionStatus, - request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, - response_serializer=InstanceCommunication__pb2.FunctionStatus.SerializeToString, - ), - 'GetAndResetMetrics': grpc.unary_unary_rpc_method_handler( - servicer.GetAndResetMetrics, - request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, - response_serializer=InstanceCommunication__pb2.MetricsData.SerializeToString, - ), - 'ResetMetrics': grpc.unary_unary_rpc_method_handler( - servicer.ResetMetrics, - request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, - response_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, - ), - 'GetMetrics': grpc.unary_unary_rpc_method_handler( - servicer.GetMetrics, - request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, - response_serializer=InstanceCommunication__pb2.MetricsData.SerializeToString, - ), - 'HealthCheck': grpc.unary_unary_rpc_method_handler( - servicer.HealthCheck, - request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, - response_serializer=InstanceCommunication__pb2.HealthCheckResult.SerializeToString, - ), - } - generic_handler = grpc.method_handlers_generic_handler( - 'proto.InstanceControl', rpc_method_handlers) - server.add_generic_rpc_handlers((generic_handler,)) + rpc_method_handlers = { + 'GetFunctionStatus': grpc.unary_unary_rpc_method_handler( + servicer.GetFunctionStatus, + request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, + response_serializer=InstanceCommunication__pb2.FunctionStatus.SerializeToString, + ), + 'GetAndResetMetrics': grpc.unary_unary_rpc_method_handler( + servicer.GetAndResetMetrics, + request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, + response_serializer=InstanceCommunication__pb2.MetricsData.SerializeToString, + ), + 'ResetMetrics': grpc.unary_unary_rpc_method_handler( + servicer.ResetMetrics, + request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, + response_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, + ), + 'GetMetrics': grpc.unary_unary_rpc_method_handler( + servicer.GetMetrics, + request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, + response_serializer=InstanceCommunication__pb2.MetricsData.SerializeToString, + ), + 'HealthCheck': grpc.unary_unary_rpc_method_handler( + servicer.HealthCheck, + request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, + response_serializer=InstanceCommunication__pb2.HealthCheckResult.SerializeToString, + ), + } + generic_handler = grpc.method_handlers_generic_handler( + 'proto.InstanceControl', rpc_method_handlers) + server.add_generic_rpc_handlers((generic_handler,)) + + + # This class is part of an EXPERIMENTAL API. +class InstanceControl(object): + """Missing associated documentation comment in .proto file.""" + + @staticmethod + def GetFunctionStatus(request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.unary_unary(request, target, '/proto.InstanceControl/GetFunctionStatus', + google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, + InstanceCommunication__pb2.FunctionStatus.FromString, + options, channel_credentials, + insecure, call_credentials, compression, wait_for_ready, timeout, metadata) + + @staticmethod + def GetAndResetMetrics(request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.unary_unary(request, target, '/proto.InstanceControl/GetAndResetMetrics', + google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, + InstanceCommunication__pb2.MetricsData.FromString, + options, channel_credentials, + insecure, call_credentials, compression, wait_for_ready, timeout, metadata) + + @staticmethod + def ResetMetrics(request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.unary_unary(request, target, '/proto.InstanceControl/ResetMetrics', + google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, + google_dot_protobuf_dot_empty__pb2.Empty.FromString, + options, channel_credentials, + insecure, call_credentials, compression, wait_for_ready, timeout, metadata) + + @staticmethod + def GetMetrics(request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.unary_unary(request, target, '/proto.InstanceControl/GetMetrics', + google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, + InstanceCommunication__pb2.MetricsData.FromString, + options, channel_credentials, + insecure, call_credentials, compression, wait_for_ready, timeout, metadata) + + @staticmethod + def HealthCheck(request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.unary_unary(request, target, '/proto.InstanceControl/HealthCheck', + google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, + InstanceCommunication__pb2.HealthCheckResult.FromString, + options, channel_credentials, + insecure, call_credentials, compression, wait_for_ready, timeout, metadata) From ebfd546c455223fbd5e07b61f1a27e150660e73f Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Fri, 20 Nov 2020 09:41:28 +0800 Subject: [PATCH 7/9] fix commets Signed-off-by: xiaolong.ran --- .../integration/functions/PulsarFunctionsTest.java | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java index aaa435b082350..d95d1d31054fd 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java @@ -26,7 +26,16 @@ import org.apache.commons.lang3.StringUtils; import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.admin.PulsarAdminException; -import org.apache.pulsar.client.api.*; +import org.apache.pulsar.client.api.BatcherBuilder; +import org.apache.pulsar.client.api.Consumer; +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.MessageId; +import org.apache.pulsar.client.api.Producer; +import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.Reader; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.SubscriptionInitialPosition; +import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.client.impl.PulsarClientImpl; import org.apache.pulsar.client.impl.schema.AvroSchema; import org.apache.pulsar.client.impl.schema.KeyValueSchema; From e9ca083c362f391d93eb40f7cf979fe0620279b1 Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Sat, 21 Nov 2020 17:03:06 +0800 Subject: [PATCH 8/9] test ci error Signed-off-by: xiaolong.ran --- pulsar-functions/instance/src/main/python/python_instance.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-functions/instance/src/main/python/python_instance.py b/pulsar-functions/instance/src/main/python/python_instance.py index 8c000f0c856fd..9129055e85d75 100644 --- a/pulsar-functions/instance/src/main/python/python_instance.py +++ b/pulsar-functions/instance/src/main/python/python_instance.py @@ -320,7 +320,7 @@ def setup_producer(self): str(self.instance_config.function_details.sink.topic), block_if_queue_full=True, batching_enabled=True, - batching_type=batch_type, + # batching_type=batch_type, batching_max_publish_delay_ms=10, compression_type=pulsar.CompressionType.LZ4, # set send timeout to be infinity to prevent potential deadlock with consumer From 1ae7762096ac98b0e7fa1ad2873a06aaff6a1370 Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Sat, 21 Nov 2020 17:25:20 +0800 Subject: [PATCH 9/9] fix ci error Signed-off-by: xiaolong.ran --- .../instance/src/main/python/python_instance.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/pulsar-functions/instance/src/main/python/python_instance.py b/pulsar-functions/instance/src/main/python/python_instance.py index 9129055e85d75..fecde7a6e8b1c 100644 --- a/pulsar-functions/instance/src/main/python/python_instance.py +++ b/pulsar-functions/instance/src/main/python/python_instance.py @@ -309,18 +309,18 @@ def setup_producer(self): len(self.instance_config.function_details.sink.topic) > 0: Log.debug("Setting up producer for topic %s" % self.instance_config.function_details.sink.topic) - batch_type = BatchingType.Default + batch_type = pulsar.BatchingType.Default if self.instance_config.function_details.sink.producerSpec.batchBuilder != None and \ len(self.instance_config.function_details.sink.producerSpec.batchBuilder) > 0: batch_builder = self.instance_config.function_details.sink.producerSpec.batchBuilder if batch_builder == "KEY_BASED": - batch_type = BatchingType.KeyBased + batch_type = pulsar.BatchingType.KeyBased self.producer = self.pulsar_client.create_producer( str(self.instance_config.function_details.sink.topic), block_if_queue_full=True, batching_enabled=True, - # batching_type=batch_type, + batching_type=batch_type, batching_max_publish_delay_ms=10, compression_type=pulsar.CompressionType.LZ4, # set send timeout to be infinity to prevent potential deadlock with consumer