1/** 2 * Licensed to the Apache Software Foundation (ASF) under one 3 * or more contributor license agreements. See the NOTICE file 4 * distributed with this work for additional information 5 * regarding copyright ownership. The ASF licenses this file 6 * to you under the Apache License, Version 2.0 (the 7 * "License"); you may not use this file except in compliance 8 * with the License. You may obtain a copy of the License at 9 * 10 * http://www.apache.org/licenses/LICENSE-2.0 11 * 12 * Unless required by applicable law or agreed to in writing, 13 * software distributed under the License is distributed on an 14 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY 15 * KIND, either express or implied. See the License for the 16 * specific language governing permissions and limitations 17 * under the License. 18 */ 19syntax = "proto2"; 20 21package pulsar.proto; 22option java_package = "org.apache.pulsar.common.api.proto"; 23option optimize_for = LITE_RUNTIME; 24 25message Schema { 26 enum Type { 27 None = 0; 28 String = 1; 29 Json = 2; 30 Protobuf = 3; 31 Avro = 4; 32 Bool = 5; 33 Int8 = 6; 34 Int16 = 7; 35 Int32 = 8; 36 Int64 = 9; 37 Float = 10; 38 Double = 11; 39 Date = 12; 40 Time = 13; 41 Timestamp = 14; 42 KeyValue = 15; 43 } 44 45 required string name = 1; 46 required bytes schema_data = 3; 47 required Type type = 4; 48 repeated KeyValue properties = 5; 49 50} 51 52message MessageIdData { 53 required uint64 ledgerId = 1; 54 required uint64 entryId = 2; 55 optional int32 partition = 3 [default = -1]; 56 optional int32 batch_index = 4 [default = -1]; 57} 58 59message KeyValue { 60 required string key = 1; 61 required string value = 2; 62} 63 64message KeyLongValue { 65 required string key = 1; 66 required uint64 value = 2; 67} 68 69message IntRange { 70 required int32 start = 1; 71 required int32 end = 2; 72} 73 74message EncryptionKeys { 75 required string key = 1; 76 required bytes value = 2; 77 repeated KeyValue metadata = 3; 78} 79 80enum CompressionType { 81 NONE = 0; 82 LZ4 = 1; 83 ZLIB = 2; 84 ZSTD = 3; 85 SNAPPY = 4; 86} 87 88message MessageMetadata { 89 required string producer_name = 1; 90 required uint64 sequence_id = 2; 91 required uint64 publish_time = 3; 92 repeated KeyValue properties = 4; 93 94 // Property set on replicated message, 95 // includes the source cluster name 96 optional string replicated_from = 5; 97 //key to decide partition for the msg 98 optional string partition_key = 6; 99 // Override namespace's replication 100 repeated string replicate_to = 7; 101 optional CompressionType compression = 8 [default = NONE]; 102 optional uint32 uncompressed_size = 9 [default = 0]; 103 // Removed below checksum field from Metadata as 104 // it should be part of send-command which keeps checksum of header + payload 105 //optional sfixed64 checksum = 10; 106 // differentiate single and batch message metadata 107 optional int32 num_messages_in_batch = 11 [default = 1]; 108 109 // the timestamp that this event occurs. it is typically set by applications. 110 // if this field is omitted, `publish_time` can be used for the purpose of `event_time`. 111 optional uint64 event_time = 12 [default = 0]; 112 // Contains encryption key name, encrypted key and metadata to describe the key 113 repeated EncryptionKeys encryption_keys = 13; 114 // Algorithm used to encrypt data key 115 optional string encryption_algo = 14; 116 // Additional parameters required by encryption 117 optional bytes encryption_param = 15; 118 optional bytes schema_version = 16; 119 120 optional bool partition_key_b64_encoded = 17 [ default = false ]; 121 // Specific a key to overwrite the message key which used for ordering dispatch in Key_Shared mode. 122 optional bytes ordering_key = 18; 123 124 // Mark the message to be delivered at or after the specified timestamp 125 optional int64 deliver_at_time = 19; 126 127 // Identify whether a message is a "marker" message used for 128 // internal metadata instead of application published data. 129 // Markers will generally not be propagated back to clients 130 optional int32 marker_type = 20; 131 132 // transaction related message info 133 optional uint64 txnid_least_bits = 22 [default = 0]; 134 optional uint64 txnid_most_bits = 23 [default = 0]; 135 136 /// Add highest sequence id to support batch message with external sequence id 137 optional uint64 highest_sequence_id = 24 [default = 0]; 138} 139 140message SingleMessageMetadata { 141 repeated KeyValue properties = 1; 142 optional string partition_key = 2; 143 required int32 payload_size = 3; 144 optional bool compacted_out = 4 [default = false]; 145 146 // the timestamp that this event occurs. it is typically set by applications. 147 // if this field is omitted, `publish_time` can be used for the purpose of `event_time`. 148 optional uint64 event_time = 5 [default = 0]; 149 optional bool partition_key_b64_encoded = 6 [ default = false ]; 150 // Specific a key to overwrite the message key which used for ordering dispatch in Key_Shared mode. 151 optional bytes ordering_key = 7; 152 // Allows consumer retrieve the sequence id that the producer set. 153 optional uint64 sequence_id = 8; 154} 155 156enum ServerError { 157 UnknownError = 0; 158 MetadataError = 1; // Error with ZK/metadata 159 PersistenceError = 2; // Error writing reading from BK 160 AuthenticationError = 3; // Non valid authentication 161 AuthorizationError = 4; // Not authorized to use resource 162 163 ConsumerBusy = 5; // Unable to subscribe/unsubscribe because 164 // other consumers are connected 165 ServiceNotReady = 6; // Any error that requires client retry operation with a fresh lookup 166 ProducerBlockedQuotaExceededError = 7; // Unable to create producer because backlog quota exceeded 167 ProducerBlockedQuotaExceededException = 8; // Exception while creating producer because quota exceeded 168 ChecksumError = 9; // Error while verifying message checksum 169 UnsupportedVersionError = 10; // Error when an older client/version doesn't support a required feature 170 TopicNotFound = 11; // Topic not found 171 SubscriptionNotFound = 12; // Subscription not found 172 ConsumerNotFound = 13; // Consumer not found 173 TooManyRequests = 14; // Error with too many simultaneously request 174 TopicTerminatedError = 15; // The topic has been terminated 175 176 ProducerBusy = 16; // Producer with same name is already connected 177 InvalidTopicName = 17; // The topic name is not valid 178 179 IncompatibleSchema = 18; // Specified schema was incompatible with topic schema 180 ConsumerAssignError = 19; // Dispatcher assign consumer error 181 182 TransactionCoordinatorNotFound = 20; // Transaction coordinator not found error 183 InvalidTxnStatus = 21; // Invalid txn status error 184} 185 186enum AuthMethod { 187 AuthMethodNone = 0; 188 AuthMethodYcaV1 = 1; 189 AuthMethodAthens = 2; 190} 191 192// Each protocol version identify new features that are 193// incrementally added to the protocol 194enum ProtocolVersion { 195 v0 = 0; // Initial versioning 196 v1 = 1; // Added application keep-alive 197 v2 = 2; // Added RedeliverUnacknowledgedMessages Command 198 v3 = 3; // Added compression with LZ4 and ZLib 199 v4 = 4; // Added batch message support 200 v5 = 5; // Added disconnect client w/o closing connection 201 v6 = 6; // Added checksum computation for metadata + payload 202 v7 = 7; // Added CommandLookupTopic - Binary Lookup 203 v8 = 8; // Added CommandConsumerStats - Client fetches broker side consumer stats 204 v9 = 9; // Added end of topic notification 205 v10 = 10;// Added proxy to broker 206 v11 = 11;// C++ consumers before this version are not correctly handling the checksum field 207 v12 = 12;// Added get topic's last messageId from broker 208 // Added CommandActiveConsumerChange 209 // Added CommandGetTopicsOfNamespace 210 v13 = 13; // Schema-registry : added avro schema format for json 211 v14 = 14; // Add CommandAuthChallenge and CommandAuthResponse for mutual auth 212 // Added Key_Shared subscription 213 v15 = 15; // Add CommandGetOrCreateSchema and CommandGetOrCreateSchemaResponse 214} 215 216message CommandConnect { 217 required string client_version = 1; 218 optional AuthMethod auth_method = 2; // Deprecated. Use "auth_method_name" instead. 219 optional string auth_method_name = 5; 220 optional bytes auth_data = 3; 221 optional int32 protocol_version = 4 [default = 0]; 222 223 // Client can ask to be proxyied to a specific broker 224 // This is only honored by a Pulsar proxy 225 optional string proxy_to_broker_url = 6; 226 227 // Original principal that was verified by 228 // a Pulsar proxy. In this case the auth info above 229 // will be the auth of the proxy itself 230 optional string original_principal = 7; 231 232 // Original auth role and auth Method that was passed 233 // to the proxy. In this case the auth info above 234 // will be the auth of the proxy itself 235 optional string original_auth_data = 8; 236 optional string original_auth_method = 9; 237 238 // Feature flags 239 optional FeatureFlags feature_flags = 10; 240} 241 242message FeatureFlags { 243 optional bool supports_auth_refresh = 1 [default = false]; 244} 245 246message CommandConnected { 247 required string server_version = 1; 248 optional int32 protocol_version = 2 [default = 0]; 249 optional int32 max_message_size = 3; 250} 251 252message CommandAuthResponse { 253 optional string client_version = 1; 254 optional AuthData response = 2; 255 optional int32 protocol_version = 3 [default = 0]; 256} 257 258message CommandAuthChallenge { 259 optional string server_version = 1; 260 optional AuthData challenge = 2; 261 optional int32 protocol_version = 3 [default = 0]; 262} 263 264// To support mutual authentication type, such as Sasl, reuse this command to mutual auth. 265message AuthData { 266 optional string auth_method_name = 1; 267 optional bytes auth_data = 2; 268} 269 270enum KeySharedMode { 271 AUTO_SPLIT = 0; 272 STICKY = 1; 273} 274 275message KeySharedMeta { 276 required KeySharedMode keySharedMode = 1; 277 repeated IntRange hashRanges = 3; 278} 279 280message CommandSubscribe { 281 enum SubType { 282 Exclusive = 0; 283 Shared = 1; 284 Failover = 2; 285 Key_Shared = 3; 286 } 287 required string topic = 1; 288 required string subscription = 2; 289 required SubType subType = 3; 290 291 required uint64 consumer_id = 4; 292 required uint64 request_id = 5; 293 optional string consumer_name = 6; 294 optional int32 priority_level = 7; 295 296 // Signal wether the subscription should be backed by a 297 // durable cursor or not 298 optional bool durable = 8 [default = true]; 299 300 // If specified, the subscription will position the cursor 301 // markd-delete position on the particular message id and 302 // will send messages from that point 303 optional MessageIdData start_message_id = 9; 304 305 /// Add optional metadata key=value to this consumer 306 repeated KeyValue metadata = 10; 307 308 optional bool read_compacted = 11; 309 310 optional Schema schema = 12; 311 enum InitialPosition { 312 Latest = 0; 313 Earliest = 1; 314 } 315 // Signal whether the subscription will initialize on latest 316 // or not -- earliest 317 optional InitialPosition initialPosition = 13 [default = Latest]; 318 319 // Mark the subscription as "replicated". Pulsar will make sure 320 // to periodically sync the state of replicated subscriptions 321 // across different clusters (when using geo-replication). 322 optional bool replicate_subscription_state = 14; 323 324 // If true, the subscribe operation will cause a topic to be 325 // created if it does not exist already (and if topic auto-creation 326 // is allowed by broker. 327 // If false, the subscribe operation will fail if the topic 328 // does not exist. 329 optional bool force_topic_creation = 15 [default = true]; 330 331 // If specified, the subscription will reset cursor's position back 332 // to specified seconds and will send messages from that point 333 optional uint64 start_message_rollback_duration_sec = 16 [default = 0]; 334 335 optional KeySharedMeta keySharedMeta = 17; 336} 337 338message CommandPartitionedTopicMetadata { 339 required string topic = 1; 340 required uint64 request_id = 2; 341 // TODO - Remove original_principal, original_auth_data, original_auth_method 342 // Original principal that was verified by 343 // a Pulsar proxy. 344 optional string original_principal = 3; 345 346 // Original auth role and auth Method that was passed 347 // to the proxy. 348 optional string original_auth_data = 4; 349 optional string original_auth_method = 5; 350} 351 352message CommandPartitionedTopicMetadataResponse { 353 enum LookupType { 354 Success = 0; 355 Failed = 1; 356 } 357 optional uint32 partitions = 1; // Optional in case of error 358 required uint64 request_id = 2; 359 optional LookupType response = 3; 360 optional ServerError error = 4; 361 optional string message = 5; 362} 363 364message CommandLookupTopic { 365 required string topic = 1; 366 required uint64 request_id = 2; 367 optional bool authoritative = 3 [default = false]; 368 369 // TODO - Remove original_principal, original_auth_data, original_auth_method 370 // Original principal that was verified by 371 // a Pulsar proxy. 372 optional string original_principal = 4; 373 374 // Original auth role and auth Method that was passed 375 // to the proxy. 376 optional string original_auth_data = 5; 377 optional string original_auth_method = 6; 378} 379 380message CommandLookupTopicResponse { 381 enum LookupType { 382 Redirect = 0; 383 Connect = 1; 384 Failed = 2; 385 } 386 387 optional string brokerServiceUrl = 1; // Optional in case of error 388 optional string brokerServiceUrlTls = 2; 389 optional LookupType response = 3; 390 required uint64 request_id = 4; 391 optional bool authoritative = 5 [default = false]; 392 optional ServerError error = 6; 393 optional string message = 7; 394 395 // If it's true, indicates to the client that it must 396 // always connect through the service url after the 397 // lookup has been completed. 398 optional bool proxy_through_service_url = 8 [default = false]; 399} 400 401/// Create a new Producer on a topic, assigning the given producer_id, 402/// all messages sent with this producer_id will be persisted on the topic 403message CommandProducer { 404 required string topic = 1; 405 required uint64 producer_id = 2; 406 required uint64 request_id = 3; 407 408 /// If a producer name is specified, the name will be used, 409 /// otherwise the broker will generate a unique name 410 optional string producer_name = 4; 411 412 optional bool encrypted = 5 [default = false]; 413 414 /// Add optional metadata key=value to this producer 415 repeated KeyValue metadata = 6; 416 417 optional Schema schema = 7; 418 419 // If producer reconnect to broker, the epoch of this producer will +1 420 optional uint64 epoch = 8 [default = 0]; 421 422 // Indicate the name of the producer is generated or user provided 423 // Use default true here is in order to be forward compatible with the client 424 optional bool user_provided_producer_name = 9 [default = true]; 425} 426 427message CommandSend { 428 required uint64 producer_id = 1; 429 required uint64 sequence_id = 2; 430 optional int32 num_messages = 3 [default = 1]; 431 optional uint64 txnid_least_bits = 4 [default = 0]; 432 optional uint64 txnid_most_bits = 5 [default = 0]; 433 434 /// Add highest sequence id to support batch message with external sequence id 435 optional uint64 highest_sequence_id = 6 [default = 0]; 436} 437 438message CommandSendReceipt { 439 required uint64 producer_id = 1; 440 required uint64 sequence_id = 2; 441 optional MessageIdData message_id = 3; 442 optional uint64 highest_sequence_id = 4 [default = 0]; 443} 444 445message CommandSendError { 446 required uint64 producer_id = 1; 447 required uint64 sequence_id = 2; 448 required ServerError error = 3; 449 required string message = 4; 450} 451 452message CommandMessage { 453 required uint64 consumer_id = 1; 454 required MessageIdData message_id = 2; 455 optional uint32 redelivery_count = 3 [default = 0]; 456} 457 458message CommandAck { 459 enum AckType { 460 Individual = 0; 461 Cumulative = 1; 462 } 463 464 required uint64 consumer_id = 1; 465 required AckType ack_type = 2; 466 467 // In case of individual acks, the client can pass a list of message ids 468 repeated MessageIdData message_id = 3; 469 470 // Acks can contain a flag to indicate the consumer 471 // received an invalid message that got discarded 472 // before being passed on to the application. 473 enum ValidationError { 474 UncompressedSizeCorruption = 0; 475 DecompressionError = 1; 476 ChecksumMismatch = 2; 477 BatchDeSerializeError = 3; 478 DecryptionError = 4; 479 } 480 481 optional ValidationError validation_error = 4; 482 repeated KeyLongValue properties = 5; 483 484 optional uint64 txnid_least_bits = 6 [default = 0]; 485 optional uint64 txnid_most_bits = 7 [default = 0]; 486} 487 488message CommandAckResponse { 489 required uint64 consumer_id = 1; 490 optional uint64 txnid_least_bits = 2 [default = 0]; 491 optional uint64 txnid_most_bits = 3 [default = 0]; 492 optional ServerError error = 4; 493 optional string message = 5; 494} 495 496// changes on active consumer 497message CommandActiveConsumerChange { 498 required uint64 consumer_id = 1; 499 optional bool is_active = 2 [default = false]; 500} 501 502message CommandFlow { 503 required uint64 consumer_id = 1; 504 505 // Max number of messages to prefetch, in addition 506 // of any number previously specified 507 required uint32 messagePermits = 2; 508} 509 510message CommandUnsubscribe { 511 required uint64 consumer_id = 1; 512 required uint64 request_id = 2; 513} 514 515// Reset an existing consumer to a particular message id 516message CommandSeek { 517 required uint64 consumer_id = 1; 518 required uint64 request_id = 2; 519 520 optional MessageIdData message_id = 3; 521 optional uint64 message_publish_time = 4; 522} 523 524// Message sent by broker to client when a topic 525// has been forcefully terminated and there are no more 526// messages left to consume 527message CommandReachedEndOfTopic { 528 required uint64 consumer_id = 1; 529} 530 531message CommandCloseProducer { 532 required uint64 producer_id = 1; 533 required uint64 request_id = 2; 534} 535 536message CommandCloseConsumer { 537 required uint64 consumer_id = 1; 538 required uint64 request_id = 2; 539} 540 541message CommandRedeliverUnacknowledgedMessages { 542 required uint64 consumer_id = 1; 543 repeated MessageIdData message_ids = 2; 544} 545 546message CommandSuccess { 547 required uint64 request_id = 1; 548 optional Schema schema = 2; 549} 550 551/// Response from CommandProducer 552message CommandProducerSuccess { 553 required uint64 request_id = 1; 554 required string producer_name = 2; 555 556 // The last sequence id that was stored by this producer in the previous session 557 // This will only be meaningful if deduplication has been enabled. 558 optional int64 last_sequence_id = 3 [default = -1]; 559 optional bytes schema_version = 4; 560} 561 562message CommandError { 563 required uint64 request_id = 1; 564 required ServerError error = 2; 565 required string message = 3; 566} 567 568// Commands to probe the state of connection. 569// When either client or broker doesn't receive commands for certain 570// amount of time, they will send a Ping probe. 571message CommandPing { 572} 573message CommandPong { 574} 575 576message CommandConsumerStats { 577 required uint64 request_id = 1; 578 // required string topic_name = 2; 579 // required string subscription_name = 3; 580 required uint64 consumer_id = 4; 581} 582 583message CommandConsumerStatsResponse { 584 required uint64 request_id = 1; 585 optional ServerError error_code = 2; 586 optional string error_message = 3; 587 588 /// Total rate of messages delivered to the consumer. msg/s 589 optional double msgRateOut = 4; 590 591 /// Total throughput delivered to the consumer. bytes/s 592 optional double msgThroughputOut = 5; 593 594 /// Total rate of messages redelivered by this consumer. msg/s 595 optional double msgRateRedeliver = 6; 596 597 /// Name of the consumer 598 optional string consumerName = 7; 599 600 /// Number of available message permits for the consumer 601 optional uint64 availablePermits = 8; 602 603 /// Number of unacknowledged messages for the consumer 604 optional uint64 unackedMessages = 9; 605 606 /// Flag to verify if consumer is blocked due to reaching threshold of unacked messages 607 optional bool blockedConsumerOnUnackedMsgs = 10; 608 609 /// Address of this consumer 610 optional string address = 11; 611 612 /// Timestamp of connection 613 optional string connectedSince = 12; 614 615 /// Whether this subscription is Exclusive or Shared or Failover 616 optional string type = 13; 617 618 /// Total rate of messages expired on this subscription. msg/s 619 optional double msgRateExpired = 14; 620 621 /// Number of messages in the subscription backlog 622 optional uint64 msgBacklog = 15; 623} 624 625message CommandGetLastMessageId { 626 required uint64 consumer_id = 1; 627 required uint64 request_id = 2; 628} 629 630message CommandGetLastMessageIdResponse { 631 required MessageIdData last_message_id = 1; 632 required uint64 request_id = 2; 633} 634 635message CommandGetTopicsOfNamespace { 636 enum Mode { 637 PERSISTENT = 0; 638 NON_PERSISTENT = 1; 639 ALL = 2; 640 } 641 required uint64 request_id = 1; 642 required string namespace = 2; 643 optional Mode mode = 3 [default = PERSISTENT]; 644} 645 646message CommandGetTopicsOfNamespaceResponse { 647 required uint64 request_id = 1; 648 repeated string topics = 2; 649} 650 651message CommandGetSchema { 652 required uint64 request_id = 1; 653 required string topic = 2; 654 655 optional bytes schema_version = 3; 656} 657 658message CommandGetSchemaResponse { 659 required uint64 request_id = 1; 660 optional ServerError error_code = 2; 661 optional string error_message = 3; 662 663 optional Schema schema = 4; 664 optional bytes schema_version = 5; 665} 666 667message CommandGetOrCreateSchema { 668 required uint64 request_id = 1; 669 required string topic = 2; 670 required Schema schema = 3; 671} 672 673message CommandGetOrCreateSchemaResponse { 674 required uint64 request_id = 1; 675 optional ServerError error_code = 2; 676 optional string error_message = 3; 677 678 optional bytes schema_version = 4; 679} 680 681/// --- transaction related --- 682 683enum TxnAction { 684 COMMIT = 0; 685 ABORT = 1; 686} 687 688message CommandNewTxn { 689 required uint64 request_id = 1; 690 optional uint64 txn_ttl_seconds = 2 [default = 0]; 691 optional uint64 tc_id = 3 [default = 0]; 692} 693 694message CommandNewTxnResponse { 695 required uint64 request_id = 1; 696 optional uint64 txnid_least_bits = 2 [default = 0]; 697 optional uint64 txnid_most_bits = 3 [default = 0]; 698 optional ServerError error = 4; 699 optional string message = 5; 700} 701 702message CommandAddPartitionToTxn { 703 required uint64 request_id = 1; 704 optional uint64 txnid_least_bits = 2 [default = 0]; 705 optional uint64 txnid_most_bits = 3 [default = 0]; 706 repeated string partitions = 4; 707} 708 709message CommandAddPartitionToTxnResponse { 710 required uint64 request_id = 1; 711 optional uint64 txnid_least_bits = 2 [default = 0]; 712 optional uint64 txnid_most_bits = 3 [default = 0]; 713 optional ServerError error = 4; 714 optional string message = 5; 715} 716 717message Subscription { 718 required string topic = 1; 719 required string subscription = 2; 720} 721message CommandAddSubscriptionToTxn { 722 required uint64 request_id = 1; 723 optional uint64 txnid_least_bits = 2 [default = 0]; 724 optional uint64 txnid_most_bits = 3 [default = 0]; 725 repeated Subscription subscription = 4; 726} 727 728message CommandAddSubscriptionToTxnResponse { 729 required uint64 request_id = 1; 730 optional uint64 txnid_least_bits = 2 [default = 0]; 731 optional uint64 txnid_most_bits = 3 [default = 0]; 732 optional ServerError error = 4; 733 optional string message = 5; 734} 735 736message CommandEndTxn { 737 required uint64 request_id = 1; 738 optional uint64 txnid_least_bits = 2 [default = 0]; 739 optional uint64 txnid_most_bits = 3 [default = 0]; 740 optional TxnAction txn_action = 4; 741} 742 743message CommandEndTxnResponse { 744 required uint64 request_id = 1; 745 optional uint64 txnid_least_bits = 2 [default = 0]; 746 optional uint64 txnid_most_bits = 3 [default = 0]; 747 optional ServerError error = 4; 748 optional string message = 5; 749} 750 751message CommandEndTxnOnPartition { 752 required uint64 request_id = 1; 753 optional uint64 txnid_least_bits = 2 [default = 0]; 754 optional uint64 txnid_most_bits = 3 [default = 0]; 755 optional string topic = 4; 756 optional TxnAction txn_action = 5; 757} 758 759message CommandEndTxnOnPartitionResponse { 760 required uint64 request_id = 1; 761 optional uint64 txnid_least_bits = 2 [default = 0]; 762 optional uint64 txnid_most_bits = 3 [default = 0]; 763 optional ServerError error = 4; 764 optional string message = 5; 765} 766 767message CommandEndTxnOnSubscription { 768 required uint64 request_id = 1; 769 optional uint64 txnid_least_bits = 2 [default = 0]; 770 optional uint64 txnid_most_bits = 3 [default = 0]; 771 optional Subscription subscription= 4; 772 optional TxnAction txn_action = 5; 773} 774 775message CommandEndTxnOnSubscriptionResponse { 776 required uint64 request_id = 1; 777 optional uint64 txnid_least_bits = 2 [default = 0]; 778 optional uint64 txnid_most_bits = 3 [default = 0]; 779 optional ServerError error = 4; 780 optional string message = 5; 781} 782 783message BaseCommand { 784 enum Type { 785 CONNECT = 2; 786 CONNECTED = 3; 787 SUBSCRIBE = 4; 788 789 PRODUCER = 5; 790 791 SEND = 6; 792 SEND_RECEIPT= 7; 793 SEND_ERROR = 8; 794 795 MESSAGE = 9; 796 ACK = 10; 797 FLOW = 11; 798 799 UNSUBSCRIBE = 12; 800 801 SUCCESS = 13; 802 ERROR = 14; 803 804 CLOSE_PRODUCER = 15; 805 CLOSE_CONSUMER = 16; 806 807 PRODUCER_SUCCESS = 17; 808 809 PING = 18; 810 PONG = 19; 811 812 REDELIVER_UNACKNOWLEDGED_MESSAGES = 20; 813 814 PARTITIONED_METADATA = 21; 815 PARTITIONED_METADATA_RESPONSE = 22; 816 817 LOOKUP = 23; 818 LOOKUP_RESPONSE = 24; 819 820 CONSUMER_STATS = 25; 821 CONSUMER_STATS_RESPONSE = 26; 822 823 REACHED_END_OF_TOPIC = 27; 824 825 SEEK = 28; 826 827 GET_LAST_MESSAGE_ID = 29; 828 GET_LAST_MESSAGE_ID_RESPONSE = 30; 829 830 ACTIVE_CONSUMER_CHANGE = 31; 831 832 833 GET_TOPICS_OF_NAMESPACE = 32; 834 GET_TOPICS_OF_NAMESPACE_RESPONSE = 33; 835 836 GET_SCHEMA = 34; 837 GET_SCHEMA_RESPONSE = 35; 838 839 AUTH_CHALLENGE = 36; 840 AUTH_RESPONSE = 37; 841 842 ACK_RESPONSE = 38; 843 844 GET_OR_CREATE_SCHEMA = 39; 845 GET_OR_CREATE_SCHEMA_RESPONSE = 40; 846 847 // transaction related 848 NEW_TXN = 50; 849 NEW_TXN_RESPONSE = 51; 850 851 ADD_PARTITION_TO_TXN = 52; 852 ADD_PARTITION_TO_TXN_RESPONSE = 53; 853 854 ADD_SUBSCRIPTION_TO_TXN = 54; 855 ADD_SUBSCRIPTION_TO_TXN_RESPONSE = 55; 856 857 END_TXN = 56; 858 END_TXN_RESPONSE = 57; 859 860 END_TXN_ON_PARTITION = 58; 861 END_TXN_ON_PARTITION_RESPONSE = 59; 862 863 END_TXN_ON_SUBSCRIPTION = 60; 864 END_TXN_ON_SUBSCRIPTION_RESPONSE = 61; 865 866 } 867 868 869 required Type type = 1; 870 871 optional CommandConnect connect = 2; 872 optional CommandConnected connected = 3; 873 874 optional CommandSubscribe subscribe = 4; 875 optional CommandProducer producer = 5; 876 optional CommandSend send = 6; 877 optional CommandSendReceipt send_receipt = 7; 878 optional CommandSendError send_error = 8; 879 optional CommandMessage message = 9; 880 optional CommandAck ack = 10; 881 optional CommandFlow flow = 11; 882 optional CommandUnsubscribe unsubscribe = 12; 883 884 optional CommandSuccess success = 13; 885 optional CommandError error = 14; 886 887 optional CommandCloseProducer close_producer = 15; 888 optional CommandCloseConsumer close_consumer = 16; 889 890 optional CommandProducerSuccess producer_success = 17; 891 optional CommandPing ping = 18; 892 optional CommandPong pong = 19; 893 optional CommandRedeliverUnacknowledgedMessages redeliverUnacknowledgedMessages = 20; 894 895 optional CommandPartitionedTopicMetadata partitionMetadata = 21; 896 optional CommandPartitionedTopicMetadataResponse partitionMetadataResponse = 22; 897 898 optional CommandLookupTopic lookupTopic = 23; 899 optional CommandLookupTopicResponse lookupTopicResponse = 24; 900 901 optional CommandConsumerStats consumerStats = 25; 902 optional CommandConsumerStatsResponse consumerStatsResponse = 26; 903 904 optional CommandReachedEndOfTopic reachedEndOfTopic = 27; 905 906 optional CommandSeek seek = 28; 907 908 optional CommandGetLastMessageId getLastMessageId = 29; 909 optional CommandGetLastMessageIdResponse getLastMessageIdResponse = 30; 910 911 optional CommandActiveConsumerChange active_consumer_change = 31; 912 913 optional CommandGetTopicsOfNamespace getTopicsOfNamespace = 32; 914 optional CommandGetTopicsOfNamespaceResponse getTopicsOfNamespaceResponse = 33; 915 916 optional CommandGetSchema getSchema = 34; 917 optional CommandGetSchemaResponse getSchemaResponse = 35; 918 919 optional CommandAuthChallenge authChallenge = 36; 920 optional CommandAuthResponse authResponse = 37; 921 922 optional CommandAckResponse ackResponse = 38; 923 924 optional CommandGetOrCreateSchema getOrCreateSchema = 39; 925 optional CommandGetOrCreateSchemaResponse getOrCreateSchemaResponse = 40; 926 927 // transaction related 928 optional CommandNewTxn newTxn = 50; 929 optional CommandNewTxnResponse newTxnResponse = 51; 930 optional CommandAddPartitionToTxn addPartitionToTxn= 52; 931 optional CommandAddPartitionToTxnResponse addPartitionToTxnResponse = 53; 932 optional CommandAddSubscriptionToTxn addSubscriptionToTxn = 54; 933 optional CommandAddSubscriptionToTxnResponse addSubscriptionToTxnResponse = 55; 934 optional CommandEndTxn endTxn = 56; 935 optional CommandEndTxnResponse endTxnResponse = 57; 936 optional CommandEndTxnOnPartition endTxnOnPartition = 58; 937 optional CommandEndTxnOnPartitionResponse endTxnOnPartitionResponse = 59; 938 optional CommandEndTxnOnSubscription endTxnOnSubscription = 60; 939 optional CommandEndTxnOnSubscriptionResponse endTxnOnSubscriptionResponse = 61; 940}