[improve][fn]: Added logLevel to change the log level of Function/Source/Sink Pod/Process to desired Level

pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/functions/FunctionConfig.java

@@ -88,6 +88,7 @@ public enum Runtime {
 
     private String outputSerdeClassName;
     private String logTopic;
+    private String logLevel;
     private ProcessingGuarantees processingGuarantees;
     // Do we want function instances to process data in the same order as in the input topics
     // This essentially means that every partition of input topic is consumed by only one instance

pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/io/SinkConfig.java

@@ -95,4 +95,5 @@ public class SinkConfig {
     private String transformFunctionClassName;
     private String transformFunctionConfig;
     private String logTopic;
+    private String logLevel;
 }

pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/io/SourceConfig.java

@@ -73,4 +73,5 @@ public class SourceConfig {
     // batchBuilder provides two types of batch construction methods, DEFAULT and KEY_BASED
     private String batchBuilder;
     private String logTopic;
+    private String logLevel;
 }

pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java

@@ -228,6 +228,9 @@ abstract class FunctionDetailsCommand extends BaseCommand {
                 + " #Java, Python, Go")
         protected String logTopic;
 
+        @Option(names = "--log-level", description = "Log level at which the logs of a Pulsar Function are produced")
+        protected String logLevel;
+
         @Option(names = {"-st", "--schema-type"}, description = "The builtin schema type or "
                 + "custom schema class name to be used for messages output by the function #Java")
         protected String schemaType = "";
@@ -506,6 +509,9 @@ void processArguments() throws Exception {
             if (null != logTopic) {
                 functionConfig.setLogTopic(logTopic);
             }
+            if (null != logLevel) {
+                functionConfig.setLogLevel(logLevel);
+            }
             if (null != className) {
                 functionConfig.setClassName(className);
             }

pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java

@@ -410,6 +410,8 @@ abstract class SinkDetailsCommand extends BaseCommand {
         protected String transformFunctionConfig;
         @Option(names = "--log-topic", description = "The topic to which the logs of a Pulsar Sink are produced")
         protected String logTopic;
+        @Option(names = "--log-level", description = "Log level at which the logs of a Pulsar Sink are produced")
+        protected String logLevel;
         @Option(names = "--runtime-flags", description = "Any flags that you want to pass to a runtime"
                 + " (for process & Kubernetes runtime only).")
         protected String runtimeFlags;
@@ -611,6 +613,9 @@ void processArguments() throws Exception {
             if (null != logTopic) {
                 sinkConfig.setLogTopic(logTopic);
             }
+            if (null != logLevel) {
+                sinkConfig.setLogLevel(logLevel);
+            }
             if (null != runtimeFlags) {
                 sinkConfig.setRuntimeFlags(runtimeFlags);
             }

pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSources.java

@@ -364,6 +364,8 @@ abstract class SourceDetailsCommand extends BaseCommand {
         protected String secretsString;
         @Option(names = "--log-topic", description = "The topic to which the logs of a Pulsar Sink are produced")
         protected String logTopic;
+        @Option(names = "--log-level", description = "Log level at which the logs of a Pulsar Source are produced")
+        protected String logLevel;
         @Option(names = "--runtime-flags", description = "Any flags that you want to pass to a runtime"
                 + " (for process & Kubernetes runtime only).")
         protected String runtimeFlags;
@@ -505,6 +507,9 @@ void processArguments() throws Exception {
             if (null != logTopic) {
                 sourceConfig.setLogTopic(logTopic);
             }
+            if (null != logLevel) {
+                sourceConfig.setLogLevel(logLevel);
+            }
             if (null != runtimeFlags) {
                 sourceConfig.setRuntimeFlags(runtimeFlags);
             }

pulsar-functions/instance/src/main/python/Function_pb2.py

@@ -39,7 +39,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\"\xa6\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\x13\n\x07\x61utoAck\x18\t \x01(\x08\x42\x02\x18\x01\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\"\xf7\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\x12\x14\n\x0cpoolMessages\x18\x08 \x01(\x08\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\"\xe5\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\x12/\n\x0f\x63ompressionType\x18\x06 \x01(\x0e\x32\x16.proto.CompressionType\"\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\"\xba\x03\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\x12H\n transformFunctionPackageLocation\x18\x07 \x01(\x0b\x32\x1e.proto.PackageLocationMetaData\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*[\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\x12\n\n\x06MANUAL\x10\x03*<\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*D\n\x0f\x43ompressionType\x12\x07\n\x03LZ4\x10\x00\x12\x08\n\x04NONE\x10\x01\x12\x08\n\x04ZLIB\x10\x02\x12\x08\n\x04ZSTD\x10\x03\x12\n\n\x06SNAPPY\x10\x04*)\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\"\xb8\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\x13\n\x07\x61utoAck\x18\t \x01(\x08\x42\x02\x18\x01\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\x12\x10\n\x08logLevel\x18\x18 \x01(\t\"\'\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\"\xf7\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\x12\x14\n\x0cpoolMessages\x18\x08 \x01(\x08\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\"\xe5\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\x12/\n\x0f\x63ompressionType\x18\x06 \x01(\x0e\x32\x16.proto.CompressionType\"\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\"\xf8\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\x12\x14\n\x0cskipToLatest\x18\x0e \x01(\x08\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\"\xba\x03\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\x12H\n transformFunctionPackageLocation\x18\x07 \x01(\x0b\x32\x1e.proto.PackageLocationMetaData\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*[\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\x12\n\n\x06MANUAL\x10\x03*<\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*D\n\x0f\x43ompressionType\x12\x07\n\x03LZ4\x10\x00\x12\x08\n\x04NONE\x10\x01\x12\x08\n\x04ZLIB\x10\x02\x12\x08\n\x04ZSTD\x10\x03\x12\n\n\x06SNAPPY\x10\x04*)\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(
@@ -575,6 +575,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='logLevel', full_name='proto.FunctionDetails.logLevel', index=23,
+      number=24, 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=[
   ],

pulsar-functions/proto/src/main/proto/Function.proto

@@ -98,6 +98,7 @@ message FunctionDetails {
     bool retainOrdering = 21;
     bool retainKeyOrdering = 22;
     SubscriptionPosition subscriptionPosition = 23;
+    string logLevel = 24;
 }
 
 message ConsumerSpec {

pulsar-functions/runtime-all/src/main/resources/java_instance_log4j2.xml

@@ -120,7 +120,7 @@
             </AppenderRef>
         </Logger>
         <Root>
-            <level>info</level>
+            <level>${sys:pulsar.log.level}</level>
             <AppenderRef>
                 <ref>${sys:pulsar.log.appender}</ref>
                 <level>${sys:pulsar.log.level}</level>

pulsar-functions/runtime-all/src/main/resources/kubernetes_instance_log4j2.xml

@@ -50,7 +50,7 @@
             </AppenderRef>
         </Logger>
         <Root>
-            <level>info</level>
+            <level>${sys:pulsar.log.level}</level>
             <AppenderRef>
                 <ref>Console</ref>
                 <level>${sys:pulsar.log.level}</level>