PulsarApi.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 = "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;
}