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

/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

syntax = "proto3";
package proto;

option java_package = "org.apache.pulsar.functions.proto";
option java_outer_classname = "Function";

enum ProcessingGuarantees {
    ATLEAST_ONCE = 0; // [default value]
    ATMOST_ONCE = 1;
    EFFECTIVELY_ONCE = 2;
    MANUAL = 3;
}

enum SubscriptionType {
    SHARED = 0;
    FAILOVER = 1;
    KEY_SHARED = 2;
}

enum SubscriptionPosition {
    LATEST = 0;
    EARLIEST = 1;
}

message Resources {
    double cpu = 1;
    int64 ram = 2;
    int64 disk = 3;
}

message RetryDetails {
    int32 maxMessageRetries = 1;
    string deadLetterTopic = 2;
}

message FunctionDetails {
    enum Runtime {
        JAVA = 0;
        PYTHON = 1;
        GO = 3;
    }
    enum ComponentType {
        UNKNOWN = 0;
        FUNCTION = 1;
        SOURCE = 2;
        SINK = 3;
    }
    string tenant = 1;
    string namespace = 2;
    string name = 3;
    string className = 4;
    string logTopic = 5;
    ProcessingGuarantees processingGuarantees = 6;
    string userConfig = 7;
    string secretsMap = 16;
    Runtime runtime = 8;
    // Deprecated since, see https://github.com/apache/pulsar/issues/15560
    bool autoAck = 9 [deprecated = true];
    int32 parallelism = 10;
    SourceSpec source = 11;
    SinkSpec sink = 12;
    Resources resources = 13;
    string packageUrl = 14; //present only if function submitted with package-url
    RetryDetails retryDetails = 15;
    string runtimeFlags = 17;
    ComponentType componentType = 18;
    string customRuntimeOptions = 19;
    /* If specified, this will refer to an archive that is
     * already present in the server */
    string builtin = 20;
    bool retainOrdering = 21;
    bool retainKeyOrdering = 22;
    SubscriptionPosition subscriptionPosition = 23;
}

message ConsumerSpec {
    string schemaType = 1;
    string serdeClassName = 2;
    bool isRegexPattern = 3;
    message ReceiverQueueSize {
        int32 value = 1;
    }
    ReceiverQueueSize receiverQueueSize = 4;
    map<string, string> schemaProperties = 5;
    map<string, string> consumerProperties = 6;
    CryptoSpec cryptoSpec = 7;
    bool poolMessages = 8;
}

message ProducerSpec {
    int32 maxPendingMessages = 1;
    int32 maxPendingMessagesAcrossPartitions = 2;
    bool useThreadLocalProducers = 3;
    CryptoSpec cryptoSpec = 4;
    string batchBuilder = 5;
}

message CryptoSpec {
    enum FailureAction {
        FAIL = 0;

        DISCARD = 1;
        CONSUME = 2;

        SEND = 10;
    }

    string cryptoKeyReaderClassName = 1;
    string cryptoKeyReaderConfig = 2;

    // key names used by producer to encrypt data
    repeated string producerEncryptionKeyName = 3;
    // define the action if producer fail to encrypt data
    // one of FAIL, SEND
    FailureAction producerCryptoFailureAction = 4;

    // define the action if consumer fail to decrypt data
    // one of FAIL, DISCARD, CONSUME
    FailureAction consumerCryptoFailureAction = 5;
}

message SourceSpec {
    string className = 1;
    // map in json format
    string configs = 2;
    string typeClassName = 5;

    // configs used only when source feeds into functions
    SubscriptionType subscriptionType = 3;

    // @deprecated -- use topicsToSchema
    map<string, string> topicsToSerDeClassName = 4 [deprecated = true];

    /**
     *
     */
    map<string, ConsumerSpec> inputSpecs = 10;

    uint64 timeoutMs = 6;
    string topicsPattern = 7 [deprecated = true];

    /* If specified, this will refer to an archive that is
     * already present in the server */
    string builtin = 8;
    string subscriptionName = 9;
    bool cleanupSubscription = 11;
    SubscriptionPosition subscriptionPosition = 12;
    uint64 negativeAckRedeliveryDelayMs = 13;
}

message SinkSpec {
    string className = 1;
    // map in json format
    string configs = 2;
    string typeClassName = 5;

    // configs used only when functions output to sink
    string topic = 3;
    ProducerSpec producerSpec = 11;

    string serDeClassName = 4;

    /* If specified, this will refer to an archive that is
     * already present in the server */
    string builtin = 6;

    /**
     * Builtin schema type or custom schema class name
     */
    string schemaType = 7;

    bool forwardSourceMessageProperty = 8;

    map<string, string> schemaProperties = 9;

    map<string, string> consumerProperties = 10;
}

message PackageLocationMetaData {
    string packagePath = 1;
    string originalFileName = 2;
}

enum FunctionState {
    RUNNING = 0;
    STOPPED = 1;
}
message FunctionMetaData {
    FunctionDetails functionDetails = 1;
    PackageLocationMetaData packageLocation = 2;
    uint64 version = 3;
    uint64 createTime = 4;
    map<int32, FunctionState> instanceStates = 5;
    FunctionAuthenticationSpec functionAuthSpec = 6;
}

message FunctionAuthenticationSpec {
    /**
     * function authentication related data that the function authentication provider
     * needs to cache/distribute to all workers support function authentication.
     * Depending on the function authentication provider implementation, this can be the actual auth credentials
     * or a pointer to the auth credentials that this function should use
     */
    bytes data = 1;
    /**
     * classname of the function auth provicer this data is relevant to
     */
    string provider = 2;
}

message Instance {
    FunctionMetaData functionMetaData = 1;
    int32 instanceId = 2;
}

message Assignment {
    Instance instance = 1;
    string workerId = 2;
}