/** * 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 = "proto2"; package pulsar.proto; option java_package = "org.apache.pulsar.common.api.proto"; option optimize_for = LITE_RUNTIME; message Schema { enum Type { None = 0; String = 1; Json = 2; Protobuf = 3; Avro = 4; Bool = 5; Int8 = 6; Int16 = 7; Int32 = 8; Int64 = 9; Float = 10; Double = 11; Date = 12; Time = 13; Timestamp = 14; KeyValue = 15; Instant = 16; LocalDate = 17; LocalTime = 18; LocalDateTime = 19; ProtobufNative = 20; } required string name = 1; required bytes schema_data = 3; required Type type = 4; repeated KeyValue properties = 5; } message MessageIdData { required uint64 ledgerId = 1; required uint64 entryId = 2; optional int32 partition = 3 [default = -1]; optional int32 batch_index = 4 [default = -1]; repeated int64 ack_set = 5; optional int32 batch_size = 6; // For the chunk message id, we need to specify the first chunk message id. optional MessageIdData first_chunk_message_id = 7; } message KeyValue { required string key = 1; required string value = 2; } message KeyLongValue { required string key = 1; required uint64 value = 2; } message IntRange { required int32 start = 1; required int32 end = 2; } message EncryptionKeys { required string key = 1; required bytes value = 2; repeated KeyValue metadata = 3; } enum CompressionType { NONE = 0; LZ4 = 1; ZLIB = 2; ZSTD = 3; SNAPPY = 4; } enum ProducerAccessMode { Shared = 0; // By default multiple producers can publish on a topic Exclusive = 1; // Require exclusive access for producer. Fail immediately if there's already a producer connected. WaitForExclusive = 2; // Producer creation is pending until it can acquire exclusive access ExclusiveWithFencing = 3; // Require exclusive access for producer. Fence out old producer. } message MessageMetadata { required string producer_name = 1; required uint64 sequence_id = 2; required uint64 publish_time = 3; repeated KeyValue properties = 4; // Property set on replicated message, // includes the source cluster name optional string replicated_from = 5; //key to decide partition for the msg optional string partition_key = 6; // Override namespace's replication repeated string replicate_to = 7; optional CompressionType compression = 8 [default = NONE]; optional uint32 uncompressed_size = 9 [default = 0]; // Removed below checksum field from Metadata as // it should be part of send-command which keeps checksum of header + payload //optional sfixed64 checksum = 10; // differentiate single and batch message metadata optional int32 num_messages_in_batch = 11 [default = 1]; // the timestamp that this event occurs. it is typically set by applications. // if this field is omitted, `publish_time` can be used for the purpose of `event_time`. optional uint64 event_time = 12 [default = 0]; // Contains encryption key name, encrypted key and metadata to describe the key repeated EncryptionKeys encryption_keys = 13; // Algorithm used to encrypt data key optional string encryption_algo = 14; // Additional parameters required by encryption optional bytes encryption_param = 15; optional bytes schema_version = 16; optional bool partition_key_b64_encoded = 17 [ default = false ]; // Specific a key to overwrite the message key which used for ordering dispatch in Key_Shared mode. optional bytes ordering_key = 18; // Mark the message to be delivered at or after the specified timestamp optional int64 deliver_at_time = 19; // Identify whether a message is a "marker" message used for // internal metadata instead of application published data. // Markers will generally not be propagated back to clients optional int32 marker_type = 20; // transaction related message info optional uint64 txnid_least_bits = 22; optional uint64 txnid_most_bits = 23; /// Add highest sequence id to support batch message with external sequence id optional uint64 highest_sequence_id = 24 [default = 0]; // Indicate if the message payload value is set optional bool null_value = 25 [default = false]; optional string uuid = 26; optional int32 num_chunks_from_msg = 27; optional int32 total_chunk_msg_size = 28; optional int32 chunk_id = 29; // Indicate if the message partition key is set optional bool null_partition_key = 30 [default = false]; } message SingleMessageMetadata { repeated KeyValue properties = 1; optional string partition_key = 2; required int32 payload_size = 3; optional bool compacted_out = 4 [default = false]; // the timestamp that this event occurs. it is typically set by applications. // if this field is omitted, `publish_time` can be used for the purpose of `event_time`. optional uint64 event_time = 5 [default = 0]; optional bool partition_key_b64_encoded = 6 [ default = false ]; // Specific a key to overwrite the message key which used for ordering dispatch in Key_Shared mode. optional bytes ordering_key = 7; // Allows consumer retrieve the sequence id that the producer set. optional uint64 sequence_id = 8; // Indicate if the message payload value is set optional bool null_value = 9 [ default = false ]; // Indicate if the message partition key is set optional bool null_partition_key = 10 [ default = false]; } // metadata added for entry from broker message BrokerEntryMetadata { optional uint64 broker_timestamp = 1; optional uint64 index = 2; } enum ServerError { UnknownError = 0; MetadataError = 1; // Error with ZK/metadata PersistenceError = 2; // Error writing reading from BK AuthenticationError = 3; // Non valid authentication AuthorizationError = 4; // Not authorized to use resource ConsumerBusy = 5; // Unable to subscribe/unsubscribe because // other consumers are connected ServiceNotReady = 6; // Any error that requires client retry operation with a fresh lookup ProducerBlockedQuotaExceededError = 7; // Unable to create producer because backlog quota exceeded ProducerBlockedQuotaExceededException = 8; // Exception while creating producer because quota exceeded ChecksumError = 9; // Error while verifying message checksum UnsupportedVersionError = 10; // Error when an older client/version doesn't support a required feature TopicNotFound = 11; // Topic not found SubscriptionNotFound = 12; // Subscription not found ConsumerNotFound = 13; // Consumer not found TooManyRequests = 14; // Error with too many simultaneously request TopicTerminatedError = 15; // The topic has been terminated ProducerBusy = 16; // Producer with same name is already connected InvalidTopicName = 17; // The topic name is not valid IncompatibleSchema = 18; // Specified schema was incompatible with topic schema ConsumerAssignError = 19; // Dispatcher assign consumer error TransactionCoordinatorNotFound = 20; // Transaction coordinator not found error InvalidTxnStatus = 21; // Invalid txn status error NotAllowedError = 22; // Not allowed error TransactionConflict = 23; // Ack with transaction conflict TransactionNotFound = 24; // Transaction not found ProducerFenced = 25; // When a producer asks and fail to get exclusive producer access, // or loses the exclusive status after a reconnection, the broker will // use this error to indicate that this producer is now permanently // fenced. Applications are now supposed to close it and create a // new producer } enum AuthMethod { AuthMethodNone = 0; AuthMethodYcaV1 = 1; AuthMethodAthens = 2; } // Each protocol version identify new features that are // incrementally added to the protocol enum ProtocolVersion { v0 = 0; // Initial versioning v1 = 1; // Added application keep-alive v2 = 2; // Added RedeliverUnacknowledgedMessages Command v3 = 3; // Added compression with LZ4 and ZLib v4 = 4; // Added batch message support v5 = 5; // Added disconnect client w/o closing connection v6 = 6; // Added checksum computation for metadata + payload v7 = 7; // Added CommandLookupTopic - Binary Lookup v8 = 8; // Added CommandConsumerStats - Client fetches broker side consumer stats v9 = 9; // Added end of topic notification v10 = 10;// Added proxy to broker v11 = 11;// C++ consumers before this version are not correctly handling the checksum field v12 = 12;// Added get topic's last messageId from broker // Added CommandActiveConsumerChange // Added CommandGetTopicsOfNamespace v13 = 13; // Schema-registry : added avro schema format for json v14 = 14; // Add CommandAuthChallenge and CommandAuthResponse for mutual auth // Added Key_Shared subscription v15 = 15; // Add CommandGetOrCreateSchema and CommandGetOrCreateSchemaResponse v16 = 16; // Add support for broker entry metadata v17 = 17; // Added support ack receipt v18 = 18; // Add client support for broker entry metadata v19 = 19; // Add CommandTcClientConnectRequest and CommandTcClientConnectResponse } message CommandConnect { required string client_version = 1; optional AuthMethod auth_method = 2; // Deprecated. Use "auth_method_name" instead. optional string auth_method_name = 5; optional bytes auth_data = 3; optional int32 protocol_version = 4 [default = 0]; // Client can ask to be proxyied to a specific broker // This is only honored by a Pulsar proxy optional string proxy_to_broker_url = 6; // Original principal that was verified by // a Pulsar proxy. In this case the auth info above // will be the auth of the proxy itself optional string original_principal = 7; // Original auth role and auth Method that was passed // to the proxy. In this case the auth info above // will be the auth of the proxy itself optional string original_auth_data = 8; optional string original_auth_method = 9; // Feature flags optional FeatureFlags feature_flags = 10; } message FeatureFlags { optional bool supports_auth_refresh = 1 [default = false]; optional bool supports_broker_entry_metadata = 2 [default = false]; optional bool supports_partial_producer = 3 [default = false]; } message CommandConnected { required string server_version = 1; optional int32 protocol_version = 2 [default = 0]; optional int32 max_message_size = 3; } message CommandAuthResponse { optional string client_version = 1; optional AuthData response = 2; optional int32 protocol_version = 3 [default = 0]; } message CommandAuthChallenge { optional string server_version = 1; optional AuthData challenge = 2; optional int32 protocol_version = 3 [default = 0]; } // To support mutual authentication type, such as Sasl, reuse this command to mutual auth. message AuthData { optional string auth_method_name = 1; optional bytes auth_data = 2; } enum KeySharedMode { AUTO_SPLIT = 0; STICKY = 1; } message KeySharedMeta { required KeySharedMode keySharedMode = 1; repeated IntRange hashRanges = 3; optional bool allowOutOfOrderDelivery = 4 [default = false]; } message CommandSubscribe { enum SubType { Exclusive = 0; Shared = 1; Failover = 2; Key_Shared = 3; } required string topic = 1; required string subscription = 2; required SubType subType = 3; required uint64 consumer_id = 4; required uint64 request_id = 5; optional string consumer_name = 6; optional int32 priority_level = 7; // Signal wether the subscription should be backed by a // durable cursor or not optional bool durable = 8 [default = true]; // If specified, the subscription will position the cursor // markd-delete position on the particular message id and // will send messages from that point optional MessageIdData start_message_id = 9; /// Add optional metadata key=value to this consumer repeated KeyValue metadata = 10; optional bool read_compacted = 11; optional Schema schema = 12; enum InitialPosition { Latest = 0; Earliest = 1; } // Signal whether the subscription will initialize on latest // or not -- earliest optional InitialPosition initialPosition = 13 [default = Latest]; // Mark the subscription as "replicated". Pulsar will make sure // to periodically sync the state of replicated subscriptions // across different clusters (when using geo-replication). optional bool replicate_subscription_state = 14; // If true, the subscribe operation will cause a topic to be // created if it does not exist already (and if topic auto-creation // is allowed by broker. // If false, the subscribe operation will fail if the topic // does not exist. optional bool force_topic_creation = 15 [default = true]; // If specified, the subscription will reset cursor's position back // to specified seconds and will send messages from that point optional uint64 start_message_rollback_duration_sec = 16 [default = 0]; optional KeySharedMeta keySharedMeta = 17; repeated KeyValue subscription_properties = 18; // The consumer epoch, when exclusive and failover consumer redeliver unack message will increase the epoch optional uint64 consumer_epoch = 19; } message CommandPartitionedTopicMetadata { required string topic = 1; required uint64 request_id = 2; // TODO - Remove original_principal, original_auth_data, original_auth_method // Original principal that was verified by // a Pulsar proxy. optional string original_principal = 3; // Original auth role and auth Method that was passed // to the proxy. optional string original_auth_data = 4; optional string original_auth_method = 5; } message CommandPartitionedTopicMetadataResponse { enum LookupType { Success = 0; Failed = 1; } optional uint32 partitions = 1; // Optional in case of error required uint64 request_id = 2; optional LookupType response = 3; optional ServerError error = 4; optional string message = 5; } message CommandLookupTopic { required string topic = 1; required uint64 request_id = 2; optional bool authoritative = 3 [default = false]; // TODO - Remove original_principal, original_auth_data, original_auth_method // Original principal that was verified by // a Pulsar proxy. optional string original_principal = 4; // Original auth role and auth Method that was passed // to the proxy. optional string original_auth_data = 5; optional string original_auth_method = 6; // optional string advertised_listener_name = 7; } message CommandLookupTopicResponse { enum LookupType { Redirect = 0; Connect = 1; Failed = 2; } optional string brokerServiceUrl = 1; // Optional in case of error optional string brokerServiceUrlTls = 2; optional LookupType response = 3; required uint64 request_id = 4; optional bool authoritative = 5 [default = false]; optional ServerError error = 6; optional string message = 7; // If it's true, indicates to the client that it must // always connect through the service url after the // lookup has been completed. optional bool proxy_through_service_url = 8 [default = false]; } /// Create a new Producer on a topic, assigning the given producer_id, /// all messages sent with this producer_id will be persisted on the topic message CommandProducer { required string topic = 1; required uint64 producer_id = 2; required uint64 request_id = 3; /// If a producer name is specified, the name will be used, /// otherwise the broker will generate a unique name optional string producer_name = 4; optional bool encrypted = 5 [default = false]; /// Add optional metadata key=value to this producer repeated KeyValue metadata = 6; optional Schema schema = 7; // If producer reconnect to broker, the epoch of this producer will +1 optional uint64 epoch = 8 [default = 0]; // Indicate the name of the producer is generated or user provided // Use default true here is in order to be forward compatible with the client optional bool user_provided_producer_name = 9 [default = true]; // Require that this producers will be the only producer allowed on the topic optional ProducerAccessMode producer_access_mode = 10 [default = Shared]; // Topic epoch is used to fence off producers that reconnects after a new // exclusive producer has already taken over. This id is assigned by the // broker on the CommandProducerSuccess. The first time, the client will // leave it empty and then it will always carry the same epoch number on // the subsequent reconnections. optional uint64 topic_epoch = 11; optional bool txn_enabled = 12 [default = false]; // Name of the initial subscription of the topic. // If this field is not set, the initial subscription will not be created. // If this field is set but the broker's `allowAutoSubscriptionCreation` // is disabled, the producer will fail to be created. optional string initial_subscription_name = 13; } message CommandSend { required uint64 producer_id = 1; required uint64 sequence_id = 2; optional int32 num_messages = 3 [default = 1]; optional uint64 txnid_least_bits = 4 [default = 0]; optional uint64 txnid_most_bits = 5 [default = 0]; /// Add highest sequence id to support batch message with external sequence id optional uint64 highest_sequence_id = 6 [default = 0]; optional bool is_chunk =7 [default = false]; // Specify if the message being published is a Pulsar marker or not optional bool marker = 8 [default = false]; } message CommandSendReceipt { required uint64 producer_id = 1; required uint64 sequence_id = 2; optional MessageIdData message_id = 3; optional uint64 highest_sequence_id = 4 [default = 0]; } message CommandSendError { required uint64 producer_id = 1; required uint64 sequence_id = 2; required ServerError error = 3; required string message = 4; } message CommandMessage { required uint64 consumer_id = 1; required MessageIdData message_id = 2; optional uint32 redelivery_count = 3 [default = 0]; repeated int64 ack_set = 4; optional uint64 consumer_epoch = 5; } message CommandAck { enum AckType { Individual = 0; Cumulative = 1; } required uint64 consumer_id = 1; required AckType ack_type = 2; // In case of individual acks, the client can pass a list of message ids repeated MessageIdData message_id = 3; // Acks can contain a flag to indicate the consumer // received an invalid message that got discarded // before being passed on to the application. enum ValidationError { UncompressedSizeCorruption = 0; DecompressionError = 1; ChecksumMismatch = 2; BatchDeSerializeError = 3; DecryptionError = 4; } optional ValidationError validation_error = 4; repeated KeyLongValue properties = 5; optional uint64 txnid_least_bits = 6 [default = 0]; optional uint64 txnid_most_bits = 7 [default = 0]; optional uint64 request_id = 8; } message CommandAckResponse { required uint64 consumer_id = 1; optional uint64 txnid_least_bits = 2 [default = 0]; optional uint64 txnid_most_bits = 3 [default = 0]; optional ServerError error = 4; optional string message = 5; optional uint64 request_id = 6; } // changes on active consumer message CommandActiveConsumerChange { required uint64 consumer_id = 1; optional bool is_active = 2 [default = false]; } message CommandFlow { required uint64 consumer_id = 1; // Max number of messages to prefetch, in addition // of any number previously specified required uint32 messagePermits = 2; } message CommandUnsubscribe { required uint64 consumer_id = 1; required uint64 request_id = 2; } // Reset an existing consumer to a particular message id message CommandSeek { required uint64 consumer_id = 1; required uint64 request_id = 2; optional MessageIdData message_id = 3; optional uint64 message_publish_time = 4; } // Message sent by broker to client when a topic // has been forcefully terminated and there are no more // messages left to consume message CommandReachedEndOfTopic { required uint64 consumer_id = 1; } message CommandCloseProducer { required uint64 producer_id = 1; required uint64 request_id = 2; } message CommandCloseConsumer { required uint64 consumer_id = 1; required uint64 request_id = 2; } message CommandRedeliverUnacknowledgedMessages { required uint64 consumer_id = 1; repeated MessageIdData message_ids = 2; optional uint64 consumer_epoch = 3; } message CommandSuccess { required uint64 request_id = 1; optional Schema schema = 2; } /// Response from CommandProducer message CommandProducerSuccess { required uint64 request_id = 1; required string producer_name = 2; // The last sequence id that was stored by this producer in the previous session // This will only be meaningful if deduplication has been enabled. optional int64 last_sequence_id = 3 [default = -1]; optional bytes schema_version = 4; // The topic epoch assigned by the broker. This field will only be set if we // were requiring exclusive access when creating the producer. optional uint64 topic_epoch = 5; // If producer is not "ready", the client will avoid to timeout the request // for creating the producer. Instead it will wait indefinitely until it gets // a subsequent `CommandProducerSuccess` with `producer_ready==true`. optional bool producer_ready = 6 [default = true]; } message CommandError { required uint64 request_id = 1; required ServerError error = 2; required string message = 3; } // Commands to probe the state of connection. // When either client or broker doesn't receive commands for certain // amount of time, they will send a Ping probe. message CommandPing { } message CommandPong { } message CommandConsumerStats { required uint64 request_id = 1; // required string topic_name = 2; // required string subscription_name = 3; required uint64 consumer_id = 4; } message CommandConsumerStatsResponse { required uint64 request_id = 1; optional ServerError error_code = 2; optional string error_message = 3; /// Total rate of messages delivered to the consumer. msg/s optional double msgRateOut = 4; /// Total throughput delivered to the consumer. bytes/s optional double msgThroughputOut = 5; /// Total rate of messages redelivered by this consumer. msg/s optional double msgRateRedeliver = 6; /// Name of the consumer optional string consumerName = 7; /// Number of available message permits for the consumer optional uint64 availablePermits = 8; /// Number of unacknowledged messages for the consumer optional uint64 unackedMessages = 9; /// Flag to verify if consumer is blocked due to reaching threshold of unacked messages optional bool blockedConsumerOnUnackedMsgs = 10; /// Address of this consumer optional string address = 11; /// Timestamp of connection optional string connectedSince = 12; /// Whether this subscription is Exclusive or Shared or Failover optional string type = 13; /// Total rate of messages expired on this subscription. msg/s optional double msgRateExpired = 14; /// Number of messages in the subscription backlog optional uint64 msgBacklog = 15; /// Total rate of messages ack. msg/s optional double messageAckRate = 16; } message CommandGetLastMessageId { required uint64 consumer_id = 1; required uint64 request_id = 2; } message CommandGetLastMessageIdResponse { required MessageIdData last_message_id = 1; required uint64 request_id = 2; optional MessageIdData consumer_mark_delete_position = 3; } message CommandGetTopicsOfNamespace { enum Mode { PERSISTENT = 0; NON_PERSISTENT = 1; ALL = 2; } required uint64 request_id = 1; required string namespace = 2; optional Mode mode = 3 [default = PERSISTENT]; optional string topics_pattern = 4; optional string topics_hash = 5; } message CommandGetTopicsOfNamespaceResponse { required uint64 request_id = 1; repeated string topics = 2; // true iff the topic list was filtered by the pattern supplied by the client optional bool filtered = 3 [default = false]; // hash computed from the names of matching topics optional string topics_hash = 4; // if false, topics is empty and the list of matching topics has not changed optional bool changed = 5 [default = true]; } message CommandGetSchema { required uint64 request_id = 1; required string topic = 2; optional bytes schema_version = 3; } message CommandGetSchemaResponse { required uint64 request_id = 1; optional ServerError error_code = 2; optional string error_message = 3; optional Schema schema = 4; optional bytes schema_version = 5; } message CommandGetOrCreateSchema { required uint64 request_id = 1; required string topic = 2; required Schema schema = 3; } message CommandGetOrCreateSchemaResponse { required uint64 request_id = 1; optional ServerError error_code = 2; optional string error_message = 3; optional bytes schema_version = 4; } /// --- transaction related --- enum TxnAction { COMMIT = 0; ABORT = 1; } message CommandTcClientConnectRequest { required uint64 request_id = 1; required uint64 tc_id = 2 [default = 0]; } message CommandTcClientConnectResponse { required uint64 request_id = 1; optional ServerError error = 2; optional string message = 3; } message CommandNewTxn { required uint64 request_id = 1; optional uint64 txn_ttl_seconds = 2 [default = 0]; optional uint64 tc_id = 3 [default = 0]; } message CommandNewTxnResponse { required uint64 request_id = 1; optional uint64 txnid_least_bits = 2 [default = 0]; optional uint64 txnid_most_bits = 3 [default = 0]; optional ServerError error = 4; optional string message = 5; } message CommandAddPartitionToTxn { required uint64 request_id = 1; optional uint64 txnid_least_bits = 2 [default = 0]; optional uint64 txnid_most_bits = 3 [default = 0]; repeated string partitions = 4; } message CommandAddPartitionToTxnResponse { required uint64 request_id = 1; optional uint64 txnid_least_bits = 2 [default = 0]; optional uint64 txnid_most_bits = 3 [default = 0]; optional ServerError error = 4; optional string message = 5; } message Subscription { required string topic = 1; required string subscription = 2; } message CommandAddSubscriptionToTxn { required uint64 request_id = 1; optional uint64 txnid_least_bits = 2 [default = 0]; optional uint64 txnid_most_bits = 3 [default = 0]; repeated Subscription subscription = 4; } message CommandAddSubscriptionToTxnResponse { required uint64 request_id = 1; optional uint64 txnid_least_bits = 2 [default = 0]; optional uint64 txnid_most_bits = 3 [default = 0]; optional ServerError error = 4; optional string message = 5; } message CommandEndTxn { required uint64 request_id = 1; optional uint64 txnid_least_bits = 2 [default = 0]; optional uint64 txnid_most_bits = 3 [default = 0]; optional TxnAction txn_action = 4; } message CommandEndTxnResponse { required uint64 request_id = 1; optional uint64 txnid_least_bits = 2 [default = 0]; optional uint64 txnid_most_bits = 3 [default = 0]; optional ServerError error = 4; optional string message = 5; } message CommandEndTxnOnPartition { required uint64 request_id = 1; optional uint64 txnid_least_bits = 2 [default = 0]; optional uint64 txnid_most_bits = 3 [default = 0]; optional string topic = 4; optional TxnAction txn_action = 5; optional uint64 txnid_least_bits_of_low_watermark = 6; } message CommandEndTxnOnPartitionResponse { required uint64 request_id = 1; optional uint64 txnid_least_bits = 2 [default = 0]; optional uint64 txnid_most_bits = 3 [default = 0]; optional ServerError error = 4; optional string message = 5; } message CommandEndTxnOnSubscription { required uint64 request_id = 1; optional uint64 txnid_least_bits = 2 [default = 0]; optional uint64 txnid_most_bits = 3 [default = 0]; optional Subscription subscription= 4; optional TxnAction txn_action = 5; optional uint64 txnid_least_bits_of_low_watermark = 6; } message CommandEndTxnOnSubscriptionResponse { required uint64 request_id = 1; optional uint64 txnid_least_bits = 2 [default = 0]; optional uint64 txnid_most_bits = 3 [default = 0]; optional ServerError error = 4; optional string message = 5; } message BaseCommand { enum Type { CONNECT = 2; CONNECTED = 3; SUBSCRIBE = 4; PRODUCER = 5; SEND = 6; SEND_RECEIPT= 7; SEND_ERROR = 8; MESSAGE = 9; ACK = 10; FLOW = 11; UNSUBSCRIBE = 12; SUCCESS = 13; ERROR = 14; CLOSE_PRODUCER = 15; CLOSE_CONSUMER = 16; PRODUCER_SUCCESS = 17; PING = 18; PONG = 19; REDELIVER_UNACKNOWLEDGED_MESSAGES = 20; PARTITIONED_METADATA = 21; PARTITIONED_METADATA_RESPONSE = 22; LOOKUP = 23; LOOKUP_RESPONSE = 24; CONSUMER_STATS = 25; CONSUMER_STATS_RESPONSE = 26; REACHED_END_OF_TOPIC = 27; SEEK = 28; GET_LAST_MESSAGE_ID = 29; GET_LAST_MESSAGE_ID_RESPONSE = 30; ACTIVE_CONSUMER_CHANGE = 31; GET_TOPICS_OF_NAMESPACE = 32; GET_TOPICS_OF_NAMESPACE_RESPONSE = 33; GET_SCHEMA = 34; GET_SCHEMA_RESPONSE = 35; AUTH_CHALLENGE = 36; AUTH_RESPONSE = 37; ACK_RESPONSE = 38; GET_OR_CREATE_SCHEMA = 39; GET_OR_CREATE_SCHEMA_RESPONSE = 40; // transaction related NEW_TXN = 50; NEW_TXN_RESPONSE = 51; ADD_PARTITION_TO_TXN = 52; ADD_PARTITION_TO_TXN_RESPONSE = 53; ADD_SUBSCRIPTION_TO_TXN = 54; ADD_SUBSCRIPTION_TO_TXN_RESPONSE = 55; END_TXN = 56; END_TXN_RESPONSE = 57; END_TXN_ON_PARTITION = 58; END_TXN_ON_PARTITION_RESPONSE = 59; END_TXN_ON_SUBSCRIPTION = 60; END_TXN_ON_SUBSCRIPTION_RESPONSE = 61; TC_CLIENT_CONNECT_REQUEST = 62; TC_CLIENT_CONNECT_RESPONSE = 63; } required Type type = 1; optional CommandConnect connect = 2; optional CommandConnected connected = 3; optional CommandSubscribe subscribe = 4; optional CommandProducer producer = 5; optional CommandSend send = 6; optional CommandSendReceipt send_receipt = 7; optional CommandSendError send_error = 8; optional CommandMessage message = 9; optional CommandAck ack = 10; optional CommandFlow flow = 11; optional CommandUnsubscribe unsubscribe = 12; optional CommandSuccess success = 13; optional CommandError error = 14; optional CommandCloseProducer close_producer = 15; optional CommandCloseConsumer close_consumer = 16; optional CommandProducerSuccess producer_success = 17; optional CommandPing ping = 18; optional CommandPong pong = 19; optional CommandRedeliverUnacknowledgedMessages redeliverUnacknowledgedMessages = 20; optional CommandPartitionedTopicMetadata partitionMetadata = 21; optional CommandPartitionedTopicMetadataResponse partitionMetadataResponse = 22; optional CommandLookupTopic lookupTopic = 23; optional CommandLookupTopicResponse lookupTopicResponse = 24; optional CommandConsumerStats consumerStats = 25; optional CommandConsumerStatsResponse consumerStatsResponse = 26; optional CommandReachedEndOfTopic reachedEndOfTopic = 27; optional CommandSeek seek = 28; optional CommandGetLastMessageId getLastMessageId = 29; optional CommandGetLastMessageIdResponse getLastMessageIdResponse = 30; optional CommandActiveConsumerChange active_consumer_change = 31; optional CommandGetTopicsOfNamespace getTopicsOfNamespace = 32; optional CommandGetTopicsOfNamespaceResponse getTopicsOfNamespaceResponse = 33; optional CommandGetSchema getSchema = 34; optional CommandGetSchemaResponse getSchemaResponse = 35; optional CommandAuthChallenge authChallenge = 36; optional CommandAuthResponse authResponse = 37; optional CommandAckResponse ackResponse = 38; optional CommandGetOrCreateSchema getOrCreateSchema = 39; optional CommandGetOrCreateSchemaResponse getOrCreateSchemaResponse = 40; // transaction related optional CommandNewTxn newTxn = 50; optional CommandNewTxnResponse newTxnResponse = 51; optional CommandAddPartitionToTxn addPartitionToTxn= 52; optional CommandAddPartitionToTxnResponse addPartitionToTxnResponse = 53; optional CommandAddSubscriptionToTxn addSubscriptionToTxn = 54; optional CommandAddSubscriptionToTxnResponse addSubscriptionToTxnResponse = 55; optional CommandEndTxn endTxn = 56; optional CommandEndTxnResponse endTxnResponse = 57; optional CommandEndTxnOnPartition endTxnOnPartition = 58; optional CommandEndTxnOnPartitionResponse endTxnOnPartitionResponse = 59; optional CommandEndTxnOnSubscription endTxnOnSubscription = 60; optional CommandEndTxnOnSubscriptionResponse endTxnOnSubscriptionResponse = 61; optional CommandTcClientConnectRequest tcClientConnectRequest = 62; optional CommandTcClientConnectResponse tcClientConnectResponse = 63; }