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;
 }