You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2025/08/29 14:42:36 UTC
(pulsar) branch master updated: [improve][doc] Cleanup some legacy PIP documents and improve PIP listing (#24684)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 4a4e692ac98 [improve][doc] Cleanup some legacy PIP documents and improve PIP listing (#24684)
4a4e692ac98 is described below
commit 4a4e692ac9843b95facfb951384dc19ca37c5432
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Fri Aug 29 17:42:30 2025 +0300
[improve][doc] Cleanup some legacy PIP documents and improve PIP listing (#24684)
---
pip/README.md | 556 +++++++-------
pip/images/pip-148/image-001-a63d31.png | Bin 0 -> 25123 bytes
pip/images/pip-148/image-002-1ee1a6.png | Bin 0 -> 26454 bytes
pip/images/pip-148/image-003-75b9f9.png | Bin 0 -> 27054 bytes
pip/images/pip-148/image-004-b09b05.png | Bin 0 -> 47851 bytes
pip/images/pip-148/image-005-bb0aee.png | Bin 0 -> 59756 bytes
pip/images/pip-148/image-006-f51a36.png | Bin 0 -> 46372 bytes
pip/images/pip-148/image-007-a14ecc.png | Bin 0 -> 36128 bytes
pip/images/pip-31/image-001-17812c.png | Bin 0 -> 10936 bytes
pip/images/pip-31/image-002-e0ba17.png | Bin 0 -> 10026 bytes
pip/images/pip-31/image-003-3cc02b.png | Bin 0 -> 18007 bytes
pip/images/pip-31/image-004-ad2169.png | Bin 0 -> 11186 bytes
pip/images/pip-39/image-001-4633a3.png | Bin 0 -> 19255 bytes
pip/images/pip-40/image-001-d0ea56.png | Bin 0 -> 48865 bytes
pip/images/pip-40/image-002-6913c4.png | Bin 0 -> 5183 bytes
pip/images/pip-40/image-003-b570dd.png | Bin 0 -> 6650 bytes
pip/images/pip-40/image-004-28764f.png | Bin 0 -> 7083 bytes
pip/images/pip-40/image-005-601b2f.png | Bin 0 -> 9508 bytes
pip/images/pip-40/image-006-d63ca4.png | Bin 0 -> 8681 bytes
pip/images/pip-40/image-007-0eb68a.png | Bin 0 -> 6994 bytes
pip/images/pip-40/image-008-29ff41.png | Bin 0 -> 8212 bytes
pip/pip-148.md | 28 +-
pip/pip-31.md | 1240 +++++++++++++------------------
pip/pip-39.md | 4 +-
pip/pip-40.md | 32 +-
25 files changed, 804 insertions(+), 1056 deletions(-)
diff --git a/pip/README.md b/pip/README.md
index 12197cf713e..7a5be02fb8d 100644
--- a/pip/README.md
+++ b/pip/README.md
@@ -118,284 +118,284 @@ The following table lists all current PIPs in this directory, sorted by PIP numb
| PIP Number | Title |
|------------|-------|
-| 1 | [Pulsar Proxy](pip-1.md) |
-| 2 | [Non Persistent topic](pip-2.md) |
-| 3 | [Message dispatch throttling](pip-3.md) |
-| 4 | [Pulsar End to End Encryption](pip-4.md) |
-| 5 | [Event time](pip-5.md) |
-| 6 | [Guaranteed Message Deduplication](pip-6.md) |
-| 7 | [Pulsar Failure domain and Anti affinity namespaces](pip-7.md) |
-| 8 | [Pulsar beyond 1M topics](pip-8.md) |
-| 9 | [Adding more Security checks to Pulsar Proxy](pip-9.md) |
-| 10 | [Remove cluster for namespace and topic names](pip-10.md) |
-| 11 | [Short topic names](pip-11.md) |
-| 12 | [Introduce builder for creating Producer Consumer Reader](pip-12.md) |
-| 13 | [Subscribe to topics represented by regular expressions](pip-13.md) |
-| 14 | [Topic compaction](pip-14.md) |
-| 15 | [Pulsar Functions](pip-15.md) |
-| 16 | [Pulsar "instance" terminology change](pip-16.md) |
-| 17 | [Tiered storage for Pulsar topics](pip-17.md) |
-| 18 | [Pulsar Replicator](pip-18.md) |
-| 19 | [Pulsar SQL](pip-19.md) |
-| 20 | [Mechanism to revoke TLS authentication](pip-20.md) |
-| 21 | [Pulsar Edge Component](pip-21.md) |
-| 22 | [Pulsar Dead Letter Topic](pip-22.md) |
-| 23 | [Message Tracing By Interceptors](pip-23.md) |
-| 24 | [Simplify memory settings](pip-24.md) |
-| 25 | [Token based authentication](pip-25.md) |
-| 26 | [Delayed Message Delivery](pip-26.md) |
-| 27 | [Add checklist in github pull request template](pip-27.md) |
-| 28 | [Pulsar Proxy Gateway Improvement](pip-28.md) |
-| 29 | [One package for both pulsar-client and pulsar-admin](pip-29.md) |
-| 30 | [change authentication provider API to support mutual authentication](pip-30.md) |
-| 31 | [Transaction Support](pip-31.md) |
-| 32 | [Go Function API, Instance and LocalRun](pip-32.md) |
-| 33 | [Replicated subscriptions](pip-33.md) |
-| 34 | [Add new subscribe type Key_shared](pip-34.md) |
-| 35 | [Improve topic lookup for topics that have high number of partitions](pip-35.md) |
-| 36 | [Max Message Size](pip-36.md) |
-| 37 | [Large message size handling in Pulsar](pip-37.md) |
-| 38 | [Batch Receiving Messages](pip-38.md) |
-| 39 | [Namespace Change Events](pip-39.md) |
-| 40 | [Pulsar Manager](pip-40.md) |
-| 41 | [Pluggable Protocol Handler](pip-41.md) |
-| 42 | [KoP - Kafka on Pulsar](pip-42.md) |
-| 43 | [producer send message with different schema](pip-43.md) |
-| 44 | [Separate schema compatibility checker for producer and consumer](pip-44.md) |
-| 45 | [Pluggable metadata interface](pip-45.md) |
-| 46 | [Next-gen Proxy](pip-46.md) |
-| 47 | [Time Based Release Plan](pip-47.md) |
-| 48 | [hierarchical admin api](pip-48.md) |
-| 49 | [Permission levels and inheritance](pip-49.md) |
-| 50 | [Package Management](pip-50.md) |
-| 51 | [Tenant policy support](pip-51.md) |
-| 52 | [Message dispatch throttling relative to publish rate](pip-52.md) |
-| 53 | [Contribute DotPulsar to Apache Pulsar](pip-53.md) |
-| 54 | [Support acknowledgement at batch index level](pip-54.md) |
-| 55 | [Refresh Authentication Credentials](pip-55.md) |
-| 56 | [Python3 Migration](pip-56.md) |
-| 57 | [Improve Broker's Zookeeper Session Timeout Handling](pip-57.md) |
-| 58 | [Support Consumers Set Custom Retry Delay](pip-58.md) |
-| 59 | [gPRC Protocol Handler](pip-59.md) |
-| 60 | [Support Proxy server with SNI routing](pip-60.md) |
-| 61 | [Advertised multiple addresses](pip-61.md) |
-| 62 | [Move connectors, adapters and Pulsar Presto to separate repositories](pip-62.md) |
-| 63 | [Readonly Topic Ownership Support](pip-63.md) |
-| 64 | [Introduce REST endpoints for producing, consuming and reading messages](pip-64.md) |
-| 65 | [Adapting Pulsar IO Sources to support Batch Sources](pip-65.md) |
-| 66 | [Pulsar Function Mesh](pip-66.md) |
-| 67 | [Pulsarctl - An alternative tools of pulsar-admin](pip-67.md) |
-| 68 | [Exclusive Producer](pip-68.md) |
-| 69 | [Schema design for Go client](pip-69.md) |
-| 70 | [Introduce lightweight broker entry metadata](pip-70.md) |
-| 71 | [Pulsar SQL migrate SchemaHandle to presto decoder](pip-71.md) |
-| 72 | [Introduce Pulsar Interface Taxonomy: Audience and Stability Classification](pip-72.md) |
-| 73 | [Configurable data source priority for message reading](pip-73.md) |
-| 74 | [Pulsar client memory limits](pip-74.md) |
-| 75 | [Replace protobuf code generator](pip-75.md) |
-| 76 | [Streaming Offload](pip-76.md) |
-| 77 | [Contribute Supernova to Apache Pulsar](pip-77.md) |
-| 78 | [Generate Docs from Code Automatically](pip-78.md) |
-| 79 | [Reduce redundant producers from partitioned producer](pip-79.md) |
-| 80 | [Unified namespace-level admin API](pip-80.md) |
-| 81 | [Split the individual acknowledgments into multiple entries](pip-81.md) |
-| 82 | [Tenant and namespace level rate limiting](pip-82.md) |
-| 83 | [Pulsar client: Message consumption with pooled buffer](pip-83.md) |
-| 84 | [Pulsar client: Redeliver command add epoch](pip-84.md) |
-| 85 | [Expose Pulsar-Client via Function/Connector BaseContext](pip-85.md) |
-| 86 | [Pulsar Functions: Preload and release external resources](pip-86.md) |
-| 87 | [Upgrade Pulsar Website Framework (Docusaurus)](pip-87.md) |
-| 88 | [Replicate schemas across multiple](pip-88.md) |
-| 89 | [Structured document logging](pip-89.md) |
-| 90 | [Expose broker entry metadata to the client](pip-90.md) |
-| 91 | [Separate lookup timeout from operation timeout](pip-91.md) |
-| 92 | [Topic policy across multiple clusters](pip-92.md) |
-| 93 | [Transaction performance tools](pip-93.md) |
-| 94 | [Message converter at broker level](pip-94.md) |
-| 95 | [Smart Listener Selection with Multiple Bind Addresses](pip-95.md) |
-| 96 | [Message payload processor for Pulsar client](pip-96.md) |
-| 97 | [Asynchronous Authentication Provider](pip-97.md) |
-| 98 | [Redesign Pulsar Information Architecture](pip-98.md) |
-| 99 | [Pulsar Proxy Extensions](pip-99.md) |
-| 100 | [Pulsar pluggable topic factory](pip-100.md) |
-| 101 | [Add seek by index feature for consumer](pip-101.md) |
-| 104 | [Add new consumer type: TableView](pip-104.md) |
-| 105 | [Support pluggable entry filter in Dispatcher](pip-105.md) |
-| 106 | [Negative acknowledgment backoff](pip-106.md) |
-| 107 | [Introduce the chunk message ID PIP](pip-107.md) |
-| 108 | [Pulsar Feature Matrix (Client and Function)](pip-108.md) |
-| 109 | [Introduce Bot to Improve Efficiency of Developing Docs](pip-109.md) |
-| 110 | [Topic metadata](pip-110.md) |
-| 111 | [Add messages produced by Protocol Handler When checking maxMessagePublishBufferSizeInMB](pip-111.md) |
-| 112 | [Generate Release Notes Automatically](pip-112.md) |
-| 116 | [Create Pulsar Writing Style Guide](pip-116.md) |
-| 117 | [Change Pulsar Standalone defaults](pip-117.md) |
-| 118 | [Do not restart brokers when ZooKeeper session expires](pip-118.md) |
-| 119 | [Enable consistent hashing by default on KeyShared dispatcher](pip-119.md) |
-| 120 | [Enable client memory limit by default](pip-120.md) |
-| 121 | [Pulsar cluster level auto failover](pip-121.md) |
-| 122 | [Change loadBalancer default loadSheddingStrategy to ThresholdShedder](pip-122.md) |
-| 123 | [Introduce Pulsar metadata CLI tool](pip-123.md) |
-| 124 | [Create init subscription before sending message to DLQ](pip-124.md) |
-| 129 | [Introduce intermediate state for ledger deletion](pip-129.md) |
-| 130 | [Apply redelivery backoff policy for ack timeout](pip-130.md) |
-| 131 | [Resolve produce chunk messages failed when topic level maxMessageSize is set](pip-131.md) |
-| 132 | [Include message header size when check maxMessageSize for non-batch message on the client side](pip-132.md) |
-| 135 | [Include MetadataStore backend for Etcd](pip-135.md) |
-| 136 | [Sync Pulsar policies across multiple clouds](pip-136.md) |
-| 137 | [Pulsar Client Shared State API](pip-137.md) |
-| 143 | [Support split bundle by specified boundaries](pip-143.md) |
-| 144 | [Making SchemaRegistry implementation configurable](pip-144.md) |
-| 146 | [ManagedCursorInfo compression](pip-146.md) |
-| 148 | [Create Pulsar client release notes](pip-148.md) |
-| 149 | [Making the REST Admin API fully async](pip-149.md) |
-| 152 | [Support subscription level dispatch rate limiter setting](pip-152.md) |
-| 154 | [Max active transaction limitation for transaction coordinator](pip-154.md) |
-| 155 | [Drop support for Python2](pip-155.md) |
-| 156 | [Build and Run Pulsar Server on Java 17](pip-156.md) |
-| 157 | [Bucketing topic metadata to allow more topics per namespace](pip-157.md) |
-| 160 | [Make transactions work more efficiently by aggregation operation for transaction log and pending ack store](pip-160.md) |
-| 161 | [Exclusive Producer: new mode ExclusiveWithFencing](pip-161.md) |
-| 162 | [LTS Releases](pip-162.md) |
-| 165 | [Auto release client useless connections](pip-165.md) |
-| 173 | [Create a built-in Function implementing the most common basic transformations](pip-173.md) |
-| 174 | [Provide new implementation for broker dispatch cache](pip-174.md) |
-| 175 | [Extend time based release process](pip-175.md) |
-| 176 | [Refactor Doc Bot](pip-176.md) |
-| 177 | [Add the classLoader field for SchemaDefinition](pip-177.md) |
-| 178 | [Multiple snapshots for transaction buffer](pip-178.md) |
-| 179 | [Support the admin API to check unknown request parameters](pip-179.md) |
-| 180 | [Shadow Topic, an alternative way to support readonly topic ownership](pip-180.md) |
-| 181 | [Pulsar Shell](pip-181.md) |
-| 182 | [Provide new load balance placement strategy implementation for ModularLoadManagerStrategy](pip-182.md) |
-| 183 | [Reduce unnecessary REST call in broker](pip-183.md) |
-| 184 | [Topic specific consumer priorityLevel](pip-184.md) |
-| 186 | [Introduce two phase deletion protocol based on system topic](pip-186.md) |
-| 187 | [Add API to analyse a subscription backlog and provide a accurate value](pip-187.md) |
-| 188 | [Cluster migration or Blue-Green cluster deployment support in Pulsar](pip-188.md) |
-| 189 | [No batching if only one message in batch](pip-189.md) |
-| 190 | [Simplify documentation release and maintenance strategy](pip-190.md) |
-| 191 | [Support batched message using entry filter](pip-191.md) |
-| 192 | [New Pulsar Broker Load Balancer](pip-192.md) |
-| 193 | [Sink preprocessing Function](pip-193.md) |
-| 194 | [Pulsar client: seek command add epoch](pip-194.md) |
-| 195 | [New bucket based delayed message tracker](pip-195.md) |
-| 198 | [Standardize PR Naming Convention using GitHub Actions](pip-198.md) |
-| 201 | [Extensions mechanism for Pulsar Admin CLI tools](pip-201.md) |
-| 204 | [Extensions for broker interceptor](pip-204.md) |
-| 205 | [Reactive Java client for Apache Pulsar](pip-205.md) |
-| 209 | [Separate C++/Python clients to own repositories](pip-209.md) |
-| 243 | [Register Jackson Java 8 support modules by default](pip-243.md) |
-| 249 | [Pulsar website redesign](pip-249.md) |
-| 259 | [Make the config httpMaxRequestHeaderSize of the pulsar web server to configurable](pip-259.md) |
-| 261 | [Restructure Getting Started section](pip-261.md) |
-| 264 | [Support OpenTelemetry metrics in Pulsar](pip-264.md) |
-| 265 | [PR-based system for managing and reviewing PIPs](pip-265.md) |
-| 275 | [Rename numWorkerThreadsForNonPersistentTopic to topicOrderedExecutorThreadNum](pip-275.md) |
-| 276 | [Add pulsar prefix to topic_load_times metric](pip-276.md) |
-| 277 | [Add current cluster marking to clusters list command](pip-277.md) |
-| 278 | [Pluggable topic compaction service](pip-278.md) |
-| 279 | [Support topic-level policies using TableView API](pip-279.md) |
-| 280 | [Refactor CLI Argument Parsing Logic for Measurement Units using JCommander's custom converter](pip-280.md) |
-| 281 | [Add notifyError method on PushSource](pip-281.md) |
-| 282 | [Add Key_Shared subscription initial position support](pip-282.md) |
-| 284 | [Migrate topic policies implementation to use TableView](pip-284.md) |
-| 286 | [Support get position based on timestamp with topic compaction](pip-286.md) |
-| 289 | [Secure Pulsar Connector Configuration](pip-289.md) |
-| 290 | [Support message encryption in WebSocket proxy](pip-290.md) |
-| 292 | [Enforce token expiration time in the Websockets plugin](pip-292.md) |
-| 293 | [Support reader to read compacted data](pip-293.md) |
-| 296 | [Add getLastMessageIds API for Reader](pip-296.md) |
-| 297 | [Support terminating Function & Connector with the fatal exception](pip-297.md) |
-| 298 | [Support read transaction buffer snapshot segments from earliest](pip-298.md) |
-| 299 | [Support setting max unacked messages at subscription level](pip-299.md) |
-| 300 | [Add RedeliverCount field to CommandAck](pip-300.md) |
-| 301 | [Separate load data storage from configuration metadata store](pip-301.md) |
-| 302 | [Support for TableView with strong read consistency](pip-302.md) |
-| 303 | [Support PartitionedTopicStats exclude publishers and subscriptions](pip-303.md) |
-| 305 | [Add OpAddEntry and pendingData statistics info in JVM metrics](pip-305.md) |
-| 306 | [Support subscribing multi topics for WebSocket](pip-306.md) |
-| 307 | [Optimize Bundle Unload(Transfer) Protocol for ExtensibleLoadManager](pip-307.md) |
-| 312 | [Use StateStoreProvider to manage state in Pulsar Functions endpoints](pip-312.md) |
-| 313 | [Support force unsubscribe using consumer api](pip-313.md) |
-| 315 | [Configurable max delay limit for delayed delivery](pip-315.md) |
-| 318 | [Don't retain null-key messages during topic compaction](pip-318.md) |
-| 320 | [OpenTelemetry Scaffolding](pip-320.md) |
-| 321 | [Split the responsibilities of namespace replication-clusters](pip-321.md) |
-| 322 | [Pulsar Rate Limiting Refactoring](pip-322.md) |
-| 323 | [Complete Backlog Quota Telemetry](pip-323.md) |
-| 324 | [Switch to Alpine Linux base Docker images](pip-324.md) |
-| 325 | [Support reading from transaction buffer for pending transaction](pip-325.md) |
-| 326 | [Create a BOM to ease dependency management](pip-326.md) |
-| 327 | [Support force topic loading for unrecoverable errors](pip-327.md) |
-| 329 | [Strategy for maintaining the latest tag to Pulsar docker images](pip-329.md) |
-| 330 | [getMessagesById gets all messages](pip-330.md) |
-| 335 | [Support Oxia metadata store plugin](pip-335.md) |
-| 337 | [SSL Factory Plugin to customize SSLContext/SSLEngine generation](pip-337.md) |
-| 339 | [Introducing the --log-topic Option for Pulsar Sinks and Sources](pip-339.md) |
-| 342 | [Support OpenTelemetry metrics in Pulsar client](pip-342.md) |
-| 343 | [Use picocli instead of jcommander](pip-343.md) |
-| 344 | [Correct the behavior of the public API pulsarClient.getPartitionsForTopic(topicName)](pip-344.md) |
-| 347 | [add role field in consumer's stat](pip-347.md) |
-| 348 | [Trigger offload on topic load stage](pip-348.md) |
-| 349 | [Add additionalSystemCursorNames ignore list for ttl check](pip-349.md) |
-| 350 | [Allow to disable the managedLedgerOffloadDeletionLagInMillis](pip-350.md) |
-| 351 | [Additional options for Pulsar-Test client to support KeyStore based TLS](pip-351.md) |
-| 352 | [Event time based topic compactor](pip-352.md) |
-| 353 | [Improve transaction message visibility for peek-messages cli](pip-353.md) |
-| 354 | [apply topK mechanism to ModularLoadManagerImpl](pip-354.md) |
-| 355 | [Enhancing Broker-Level Metrics for Pulsar](pip-355.md) |
-| 356 | [Support Geo-Replication starts at earliest position](pip-356.md) |
-| 357 | [Correct the conf name in load balance module.](pip-357.md) |
-| 358 | [let resource weight work for OverloadShedder, LeastLongTermMessageRate, ModularLoadManagerImpl.](pip-358.md) |
-| 359 | [Support custom message listener executor for specific subscription](pip-359.md) |
-| 360 | [Admin API to display Schema metadata](pip-360.md) |
-| 363 | [Add callback parameters to the method: `org.apache.pulsar.client.impl.SendCallback.sendComplete`.](pip-363.md) |
-| 364 | [Introduce a new load balance algorithm AvgShedder](pip-364.md) |
-| 366 | [Support to specify different config for Configuration and Local Metadata Store](pip-366.md) |
-| 367 | [Propose a Contributor Repository for Pulsar](pip-367.md) |
-| 368 | [Support lookup based on the lookup properties](pip-368.md) |
-| 369 | [Flag based selective unload on changing ns-isolation-policy](pip-369.md) |
-| 370 | [configurable remote topic creation in geo-replication](pip-370.md) |
-| 373 | [Add a topic's system prop that indicates whether users have published TXN messages in before.](pip-373.md) |
-| 374 | [Visibility of messages in receiverQueue for the consumers](pip-374.md) |
-| 376 | [Make Topic Policies Service Pluggable](pip-376.md) |
-| 378 | [Add ServiceUnitStateTableView abstraction (ExtensibleLoadMangerImpl only)](pip-378.md) |
-| 379 | [Key_Shared Draining Hashes for Improved Message Ordering](pip-379.md) |
-| 380 | [Support setting up specific namespaces to skipping the load-shedding](pip-380.md) |
-| 381 | [Handle large PositionInfo state](pip-381.md) |
-| 383 | [Support granting/revoking permissions for multiple topics](pip-383.md) |
-| 384 | [ManagedLedger interface decoupling](pip-384.md) |
-| 389 | [Add Producer config compressMinMsgBodySize to improve compression performance](pip-389.md) |
-| 391 | [Improve Batch Messages Acknowledgment](pip-391.md) |
-| 392 | [Add configuration to enable consistent hashing to select active consumer for partitioned topic](pip-392.md) |
-| 393 | [Improve performance of Negative Acknowledgement](pip-393.md) |
-| 395 | [Add Proxy configuration to support configurable response headers for http reverse-proxy](pip-395.md) |
-| 396 | [Align WindowFunction's WindowContext with BaseContext](pip-396.md) |
-| 399 | [Fix Metric Name for Delayed Queue](pip-399.md) |
-| 401 | [Support set batching configurations for Pulsar Functions&Sources](pip-401.md) |
-| 402 | [Role Anonymizer for Pulsar Logging](pip-402.md) |
-| 404 | [Introduce per-ledger properties](pip-404.md) |
-| 406 | [Introduce metrics related to dispatch throttled events](pip-406.md) |
-| 407 | [Add a newMessage API to create a message with a schema and transaction](pip-407.md) |
-| 409 | [support producer configuration for retry/dead letter topic producer](pip-409.md) |
-| 412 | [Support setting messagePayloadProcessor in Pulsar Functions and Sinks](pip-412.md) |
-| 414 | [Enforce topic consistency check](pip-414.md) |
-| 415 | [Support getting message ID by index](pip-415.md) |
-| 416 | [Add a new topic method to implement trigger offload by size threshold](pip-416.md) |
-| 420 | [Provides an ability for Pulsar clients to integrate with third-party schema registry service](pip-420.md) |
-| 421 | [Require Java 17 as the minimum for Pulsar Java client SDK](pip-421.md) |
-| 422 | [Support global topic-level policy: replicated clusters and new API to delete topic-level policies](pip-422.md) |
-| 425 | [Support connecting with next available endpoint for multi-endpoint serviceUrls](pip-425.md) |
-| 427 | [Align pulsar-admin Default for Mark-Delete Rate with Broker Configuration](pip-427.md) |
-| 428 | [Change TopicPoliciesService interface to fix consistency issues](pip-428.md) |
-| 429 | [Optimize Handling of Compacted Last Entry by Skipping Payload Buffer Parsing](pip-429.md) |
-| 430 | [Pulsar Broker cache improvements: refactoring eviction and adding a new cache strategy based on expected read count](pip-430.md) |
-| 431 | [Add Creation and Last Publish Timestamps to Topic Stats](pip-431.md) |
-| 432 | [Add isEncrypted field to EncryptionContext](pip-432.md) |
-| 433 | [Optimize the conflicts of the replication and automatic creation mechanisms, including the automatic creation of topics and schemas](pip-433.md) |
-| 435 | [Add `startTimestamp` and `endTimestamp` for consuming messages in client cli](pip-435.md) |
-| 436 | [Add decryptFailListener to Consumer](pip-436.md) |
+| PIP-1 | [Pulsar Proxy](pip-1.md) |
+| PIP-2 | [Non Persistent topic](pip-2.md) |
+| PIP-3 | [Message dispatch throttling](pip-3.md) |
+| PIP-4 | [Pulsar End to End Encryption](pip-4.md) |
+| PIP-5 | [Event time](pip-5.md) |
+| PIP-6 | [Guaranteed Message Deduplication](pip-6.md) |
+| PIP-7 | [Pulsar Failure domain and Anti affinity namespaces](pip-7.md) |
+| PIP-8 | [Pulsar beyond 1M topics](pip-8.md) |
+| PIP-9 | [Adding more Security checks to Pulsar Proxy](pip-9.md) |
+| PIP-10 | [Remove cluster for namespace and topic names](pip-10.md) |
+| PIP-11 | [Short topic names](pip-11.md) |
+| PIP-12 | [Introduce builder for creating Producer Consumer Reader](pip-12.md) |
+| PIP-13 | [Subscribe to topics represented by regular expressions](pip-13.md) |
+| PIP-14 | [Topic compaction](pip-14.md) |
+| PIP-15 | [Pulsar Functions](pip-15.md) |
+| PIP-16 | [Pulsar "instance" terminology change](pip-16.md) |
+| PIP-17 | [Tiered storage for Pulsar topics](pip-17.md) |
+| PIP-18 | [Pulsar Replicator](pip-18.md) |
+| PIP-19 | [Pulsar SQL](pip-19.md) |
+| PIP-20 | [Mechanism to revoke TLS authentication](pip-20.md) |
+| PIP-21 | [Pulsar Edge Component](pip-21.md) |
+| PIP-22 | [Pulsar Dead Letter Topic](pip-22.md) |
+| PIP-23 | [Message Tracing By Interceptors](pip-23.md) |
+| PIP-24 | [Simplify memory settings](pip-24.md) |
+| PIP-25 | [Token based authentication](pip-25.md) |
+| PIP-26 | [Delayed Message Delivery](pip-26.md) |
+| PIP-27 | [Add checklist in github pull request template](pip-27.md) |
+| PIP-28 | [Pulsar Proxy Gateway Improvement](pip-28.md) |
+| PIP-29 | [One package for both pulsar-client and pulsar-admin](pip-29.md) |
+| PIP-30 | [change authentication provider API to support mutual authentication](pip-30.md) |
+| PIP-31 | [Transaction Support](pip-31.md) |
+| PIP-32 | [Go Function API, Instance and LocalRun](pip-32.md) |
+| PIP-33 | [Replicated subscriptions](pip-33.md) |
+| PIP-34 | [Add new subscribe type Key_shared](pip-34.md) |
+| PIP-35 | [Improve topic lookup for topics that have high number of partitions](pip-35.md) |
+| PIP-36 | [Max Message Size](pip-36.md) |
+| PIP-37 | [Large message size handling in Pulsar](pip-37.md) |
+| PIP-38 | [Batch Receiving Messages](pip-38.md) |
+| PIP-39 | [Namespace Change Events](pip-39.md) |
+| PIP-40 | [Pulsar Manager](pip-40.md) |
+| PIP-41 | [Pluggable Protocol Handler](pip-41.md) |
+| PIP-42 | [KoP - Kafka on Pulsar](pip-42.md) |
+| PIP-43 | [producer send message with different schema](pip-43.md) |
+| PIP-44 | [Separate schema compatibility checker for producer and consumer](pip-44.md) |
+| PIP-45 | [Pluggable metadata interface](pip-45.md) |
+| PIP-46 | [Next-gen Proxy](pip-46.md) |
+| PIP-47 | [Time Based Release Plan](pip-47.md) |
+| PIP-48 | [hierarchical admin api](pip-48.md) |
+| PIP-49 | [Permission levels and inheritance](pip-49.md) |
+| PIP-50 | [Package Management](pip-50.md) |
+| PIP-51 | [Tenant policy support](pip-51.md) |
+| PIP-52 | [Message dispatch throttling relative to publish rate](pip-52.md) |
+| PIP-53 | [Contribute DotPulsar to Apache Pulsar](pip-53.md) |
+| PIP-54 | [Support acknowledgement at batch index level](pip-54.md) |
+| PIP-55 | [Refresh Authentication Credentials](pip-55.md) |
+| PIP-56 | [Python3 Migration](pip-56.md) |
+| PIP-57 | [Improve Broker's Zookeeper Session Timeout Handling](pip-57.md) |
+| PIP-58 | [Support Consumers Set Custom Retry Delay](pip-58.md) |
+| PIP-59 | [gPRC Protocol Handler](pip-59.md) |
+| PIP-60 | [Support Proxy server with SNI routing](pip-60.md) |
+| PIP-61 | [Advertised multiple addresses](pip-61.md) |
+| PIP-62 | [Move connectors, adapters and Pulsar Presto to separate repositories](pip-62.md) |
+| PIP-63 | [Readonly Topic Ownership Support](pip-63.md) |
+| PIP-64 | [Introduce REST endpoints for producing, consuming and reading messages](pip-64.md) |
+| PIP-65 | [Adapting Pulsar IO Sources to support Batch Sources](pip-65.md) |
+| PIP-66 | [Pulsar Function Mesh](pip-66.md) |
+| PIP-67 | [Pulsarctl - An alternative tools of pulsar-admin](pip-67.md) |
+| PIP-68 | [Exclusive Producer](pip-68.md) |
+| PIP-69 | [Schema design for Go client](pip-69.md) |
+| PIP-70 | [Introduce lightweight broker entry metadata](pip-70.md) |
+| PIP-71 | [Pulsar SQL migrate SchemaHandle to presto decoder](pip-71.md) |
+| PIP-72 | [Introduce Pulsar Interface Taxonomy: Audience and Stability Classification](pip-72.md) |
+| PIP-73 | [Configurable data source priority for message reading](pip-73.md) |
+| PIP-74 | [Pulsar client memory limits](pip-74.md) |
+| PIP-75 | [Replace protobuf code generator](pip-75.md) |
+| PIP-76 | [Streaming Offload](pip-76.md) |
+| PIP-77 | [Contribute Supernova to Apache Pulsar](pip-77.md) |
+| PIP-78 | [Generate Docs from Code Automatically](pip-78.md) |
+| PIP-79 | [Reduce redundant producers from partitioned producer](pip-79.md) |
+| PIP-80 | [Unified namespace-level admin API](pip-80.md) |
+| PIP-81 | [Split the individual acknowledgments into multiple entries](pip-81.md) |
+| PIP-82 | [Tenant and namespace level rate limiting](pip-82.md) |
+| PIP-83 | [Pulsar client: Message consumption with pooled buffer](pip-83.md) |
+| PIP-84 | [Pulsar client: Redeliver command add epoch](pip-84.md) |
+| PIP-85 | [Expose Pulsar-Client via Function/Connector BaseContext](pip-85.md) |
+| PIP-86 | [Pulsar Functions: Preload and release external resources](pip-86.md) |
+| PIP-87 | [Upgrade Pulsar Website Framework (Docusaurus)](pip-87.md) |
+| PIP-88 | [Replicate schemas across multiple](pip-88.md) |
+| PIP-89 | [Structured document logging](pip-89.md) |
+| PIP-90 | [Expose broker entry metadata to the client](pip-90.md) |
+| PIP-91 | [Separate lookup timeout from operation timeout](pip-91.md) |
+| PIP-92 | [Topic policy across multiple clusters](pip-92.md) |
+| PIP-93 | [Transaction performance tools](pip-93.md) |
+| PIP-94 | [Message converter at broker level](pip-94.md) |
+| PIP-95 | [Smart Listener Selection with Multiple Bind Addresses](pip-95.md) |
+| PIP-96 | [Message payload processor for Pulsar client](pip-96.md) |
+| PIP-97 | [Asynchronous Authentication Provider](pip-97.md) |
+| PIP-98 | [Redesign Pulsar Information Architecture](pip-98.md) |
+| PIP-99 | [Pulsar Proxy Extensions](pip-99.md) |
+| PIP-100 | [Pulsar pluggable topic factory](pip-100.md) |
+| PIP-101 | [Add seek by index feature for consumer](pip-101.md) |
+| PIP-104 | [Add new consumer type: TableView](pip-104.md) |
+| PIP-105 | [Support pluggable entry filter in Dispatcher](pip-105.md) |
+| PIP-106 | [Negative acknowledgment backoff](pip-106.md) |
+| PIP-107 | [Introduce the chunk message ID PIP](pip-107.md) |
+| PIP-108 | [Pulsar Feature Matrix (Client and Function)](pip-108.md) |
+| PIP-109 | [Introduce Bot to Improve Efficiency of Developing Docs](pip-109.md) |
+| PIP-110 | [Topic metadata](pip-110.md) |
+| PIP-111 | [Add messages produced by Protocol Handler When checking maxMessagePublishBufferSizeInMB](pip-111.md) |
+| PIP-112 | [Generate Release Notes Automatically](pip-112.md) |
+| PIP-116 | [Create Pulsar Writing Style Guide](pip-116.md) |
+| PIP-117 | [Change Pulsar Standalone defaults](pip-117.md) |
+| PIP-118 | [Do not restart brokers when ZooKeeper session expires](pip-118.md) |
+| PIP-119 | [Enable consistent hashing by default on KeyShared dispatcher](pip-119.md) |
+| PIP-120 | [Enable client memory limit by default](pip-120.md) |
+| PIP-121 | [Pulsar cluster level auto failover](pip-121.md) |
+| PIP-122 | [Change loadBalancer default loadSheddingStrategy to ThresholdShedder](pip-122.md) |
+| PIP-123 | [Introduce Pulsar metadata CLI tool](pip-123.md) |
+| PIP-124 | [Create init subscription before sending message to DLQ](pip-124.md) |
+| PIP-129 | [Introduce intermediate state for ledger deletion](pip-129.md) |
+| PIP-130 | [Apply redelivery backoff policy for ack timeout](pip-130.md) |
+| PIP-131 | [Resolve produce chunk messages failed when topic level maxMessageSize is set](pip-131.md) |
+| PIP-132 | [Include message header size when check maxMessageSize for non-batch message on the client side](pip-132.md) |
+| PIP-135 | [Include MetadataStore backend for Etcd](pip-135.md) |
+| PIP-136 | [Sync Pulsar policies across multiple clouds](pip-136.md) |
+| PIP-137 | [Pulsar Client Shared State API](pip-137.md) |
+| PIP-143 | [Support split bundle by specified boundaries](pip-143.md) |
+| PIP-144 | [Making SchemaRegistry implementation configurable](pip-144.md) |
+| PIP-146 | [ManagedCursorInfo compression](pip-146.md) |
+| PIP-148 | [Create Pulsar client release notes](pip-148.md) |
+| PIP-149 | [Making the REST Admin API fully async](pip-149.md) |
+| PIP-152 | [Support subscription level dispatch rate limiter setting](pip-152.md) |
+| PIP-154 | [Max active transaction limitation for transaction coordinator](pip-154.md) |
+| PIP-155 | [Drop support for Python2](pip-155.md) |
+| PIP-156 | [Build and Run Pulsar Server on Java 17](pip-156.md) |
+| PIP-157 | [Bucketing topic metadata to allow more topics per namespace](pip-157.md) |
+| PIP-160 | [Make transactions work more efficiently by aggregation operation for transaction log and pending ack store](pip-160.md) |
+| PIP-161 | [Exclusive Producer: new mode ExclusiveWithFencing](pip-161.md) |
+| PIP-162 | [LTS Releases](pip-162.md) |
+| PIP-165 | [Auto release client useless connections](pip-165.md) |
+| PIP-173 | [Create a built-in Function implementing the most common basic transformations](pip-173.md) |
+| PIP-174 | [Provide new implementation for broker dispatch cache](pip-174.md) |
+| PIP-175 | [Extend time based release process](pip-175.md) |
+| PIP-176 | [Refactor Doc Bot](pip-176.md) |
+| PIP-177 | [Add the classLoader field for SchemaDefinition](pip-177.md) |
+| PIP-178 | [Multiple snapshots for transaction buffer](pip-178.md) |
+| PIP-179 | [Support the admin API to check unknown request parameters](pip-179.md) |
+| PIP-180 | [Shadow Topic, an alternative way to support readonly topic ownership](pip-180.md) |
+| PIP-181 | [Pulsar Shell](pip-181.md) |
+| PIP-182 | [Provide new load balance placement strategy implementation for ModularLoadManagerStrategy](pip-182.md) |
+| PIP-183 | [Reduce unnecessary REST call in broker](pip-183.md) |
+| PIP-184 | [Topic specific consumer priorityLevel](pip-184.md) |
+| PIP-186 | [Introduce two phase deletion protocol based on system topic](pip-186.md) |
+| PIP-187 | [Add API to analyse a subscription backlog and provide a accurate value](pip-187.md) |
+| PIP-188 | [Cluster migration or Blue-Green cluster deployment support in Pulsar](pip-188.md) |
+| PIP-189 | [No batching if only one message in batch](pip-189.md) |
+| PIP-190 | [Simplify documentation release and maintenance strategy](pip-190.md) |
+| PIP-191 | [Support batched message using entry filter](pip-191.md) |
+| PIP-192 | [New Pulsar Broker Load Balancer](pip-192.md) |
+| PIP-193 | [Sink preprocessing Function](pip-193.md) |
+| PIP-194 | [Pulsar client: seek command add epoch](pip-194.md) |
+| PIP-195 | [New bucket based delayed message tracker](pip-195.md) |
+| PIP-198 | [Standardize PR Naming Convention using GitHub Actions](pip-198.md) |
+| PIP-201 | [Extensions mechanism for Pulsar Admin CLI tools](pip-201.md) |
+| PIP-204 | [Extensions for broker interceptor](pip-204.md) |
+| PIP-205 | [Reactive Java client for Apache Pulsar](pip-205.md) |
+| PIP-209 | [Separate C++/Python clients to own repositories](pip-209.md) |
+| PIP-243 | [Register Jackson Java 8 support modules by default](pip-243.md) |
+| PIP-249 | [Pulsar website redesign](pip-249.md) |
+| PIP-259 | [Make the config httpMaxRequestHeaderSize of the pulsar web server to configurable](pip-259.md) |
+| PIP-261 | [Restructure Getting Started section](pip-261.md) |
+| PIP-264 | [Support OpenTelemetry metrics in Pulsar](pip-264.md) |
+| PIP-265 | [PR-based system for managing and reviewing PIPs](pip-265.md) |
+| PIP-275 | [Rename numWorkerThreadsForNonPersistentTopic to topicOrderedExecutorThreadNum](pip-275.md) |
+| PIP-276 | [Add pulsar prefix to topic_load_times metric](pip-276.md) |
+| PIP-277 | [Add current cluster marking to clusters list command](pip-277.md) |
+| PIP-278 | [Pluggable topic compaction service](pip-278.md) |
+| PIP-279 | [Support topic-level policies using TableView API](pip-279.md) |
+| PIP-280 | [Refactor CLI Argument Parsing Logic for Measurement Units using JCommander's custom converter](pip-280.md) |
+| PIP-281 | [Add notifyError method on PushSource](pip-281.md) |
+| PIP-282 | [Add Key_Shared subscription initial position support](pip-282.md) |
+| PIP-284 | [Migrate topic policies implementation to use TableView](pip-284.md) |
+| PIP-286 | [Support get position based on timestamp with topic compaction](pip-286.md) |
+| PIP-289 | [Secure Pulsar Connector Configuration](pip-289.md) |
+| PIP-290 | [Support message encryption in WebSocket proxy](pip-290.md) |
+| PIP-292 | [Enforce token expiration time in the Websockets plugin](pip-292.md) |
+| PIP-293 | [Support reader to read compacted data](pip-293.md) |
+| PIP-296 | [Add getLastMessageIds API for Reader](pip-296.md) |
+| PIP-297 | [Support terminating Function & Connector with the fatal exception](pip-297.md) |
+| PIP-298 | [Support read transaction buffer snapshot segments from earliest](pip-298.md) |
+| PIP-299 | [Support setting max unacked messages at subscription level](pip-299.md) |
+| PIP-300 | [Add RedeliverCount field to CommandAck](pip-300.md) |
+| PIP-301 | [Separate load data storage from configuration metadata store](pip-301.md) |
+| PIP-302 | [Support for TableView with strong read consistency](pip-302.md) |
+| PIP-303 | [Support PartitionedTopicStats exclude publishers and subscriptions](pip-303.md) |
+| PIP-305 | [Add OpAddEntry and pendingData statistics info in JVM metrics](pip-305.md) |
+| PIP-306 | [Support subscribing multi topics for WebSocket](pip-306.md) |
+| PIP-307 | [Optimize Bundle Unload(Transfer) Protocol for ExtensibleLoadManager](pip-307.md) |
+| PIP-312 | [Use StateStoreProvider to manage state in Pulsar Functions endpoints](pip-312.md) |
+| PIP-313 | [Support force unsubscribe using consumer api](pip-313.md) |
+| PIP-315 | [Configurable max delay limit for delayed delivery](pip-315.md) |
+| PIP-318 | [Don't retain null-key messages during topic compaction](pip-318.md) |
+| PIP-320 | [OpenTelemetry Scaffolding](pip-320.md) |
+| PIP-321 | [Split the responsibilities of namespace replication-clusters](pip-321.md) |
+| PIP-322 | [Pulsar Rate Limiting Refactoring](pip-322.md) |
+| PIP-323 | [Complete Backlog Quota Telemetry](pip-323.md) |
+| PIP-324 | [Switch to Alpine Linux base Docker images](pip-324.md) |
+| PIP-325 | [Support reading from transaction buffer for pending transaction](pip-325.md) |
+| PIP-326 | [Create a BOM to ease dependency management](pip-326.md) |
+| PIP-327 | [Support force topic loading for unrecoverable errors](pip-327.md) |
+| PIP-329 | [Strategy for maintaining the latest tag to Pulsar docker images](pip-329.md) |
+| PIP-330 | [getMessagesById gets all messages](pip-330.md) |
+| PIP-335 | [Support Oxia metadata store plugin](pip-335.md) |
+| PIP-337 | [SSL Factory Plugin to customize SSLContext/SSLEngine generation](pip-337.md) |
+| PIP-339 | [Introducing the --log-topic Option for Pulsar Sinks and Sources](pip-339.md) |
+| PIP-342 | [Support OpenTelemetry metrics in Pulsar client](pip-342.md) |
+| PIP-343 | [Use picocli instead of jcommander](pip-343.md) |
+| PIP-344 | [Correct the behavior of the public API pulsarClient.getPartitionsForTopic(topicName)](pip-344.md) |
+| PIP-347 | [add role field in consumer's stat](pip-347.md) |
+| PIP-348 | [Trigger offload on topic load stage](pip-348.md) |
+| PIP-349 | [Add additionalSystemCursorNames ignore list for ttl check](pip-349.md) |
+| PIP-350 | [Allow to disable the managedLedgerOffloadDeletionLagInMillis](pip-350.md) |
+| PIP-351 | [Additional options for Pulsar-Test client to support KeyStore based TLS](pip-351.md) |
+| PIP-352 | [Event time based topic compactor](pip-352.md) |
+| PIP-353 | [Improve transaction message visibility for peek-messages cli](pip-353.md) |
+| PIP-354 | [apply topK mechanism to ModularLoadManagerImpl](pip-354.md) |
+| PIP-355 | [Enhancing Broker-Level Metrics for Pulsar](pip-355.md) |
+| PIP-356 | [Support Geo-Replication starts at earliest position](pip-356.md) |
+| PIP-357 | [Correct the conf name in load balance module.](pip-357.md) |
+| PIP-358 | [let resource weight work for OverloadShedder, LeastLongTermMessageRate, ModularLoadManagerImpl.](pip-358.md) |
+| PIP-359 | [Support custom message listener executor for specific subscription](pip-359.md) |
+| PIP-360 | [Admin API to display Schema metadata](pip-360.md) |
+| PIP-363 | [Add callback parameters to the method: `org.apache.pulsar.client.impl.SendCallback.sendComplete`.](pip-363.md) |
+| PIP-364 | [Introduce a new load balance algorithm AvgShedder](pip-364.md) |
+| PIP-366 | [Support to specify different config for Configuration and Local Metadata Store](pip-366.md) |
+| PIP-367 | [Propose a Contributor Repository for Pulsar](pip-367.md) |
+| PIP-368 | [Support lookup based on the lookup properties](pip-368.md) |
+| PIP-369 | [Flag based selective unload on changing ns-isolation-policy](pip-369.md) |
+| PIP-370 | [configurable remote topic creation in geo-replication](pip-370.md) |
+| PIP-373 | [Add a topic's system prop that indicates whether users have published TXN messages in before.](pip-373.md) |
+| PIP-374 | [Visibility of messages in receiverQueue for the consumers](pip-374.md) |
+| PIP-376 | [Make Topic Policies Service Pluggable](pip-376.md) |
+| PIP-378 | [Add ServiceUnitStateTableView abstraction (ExtensibleLoadMangerImpl only)](pip-378.md) |
+| PIP-379 | [Key_Shared Draining Hashes for Improved Message Ordering](pip-379.md) |
+| PIP-380 | [Support setting up specific namespaces to skipping the load-shedding](pip-380.md) |
+| PIP-381 | [Handle large PositionInfo state](pip-381.md) |
+| PIP-383 | [Support granting/revoking permissions for multiple topics](pip-383.md) |
+| PIP-384 | [ManagedLedger interface decoupling](pip-384.md) |
+| PIP-389 | [Add Producer config compressMinMsgBodySize to improve compression performance](pip-389.md) |
+| PIP-391 | [Improve Batch Messages Acknowledgment](pip-391.md) |
+| PIP-392 | [Add configuration to enable consistent hashing to select active consumer for partitioned topic](pip-392.md) |
+| PIP-393 | [Improve performance of Negative Acknowledgement](pip-393.md) |
+| PIP-395 | [Add Proxy configuration to support configurable response headers for http reverse-proxy](pip-395.md) |
+| PIP-396 | [Align WindowFunction's WindowContext with BaseContext](pip-396.md) |
+| PIP-399 | [Fix Metric Name for Delayed Queue](pip-399.md) |
+| PIP-401 | [Support set batching configurations for Pulsar Functions&Sources](pip-401.md) |
+| PIP-402 | [Role Anonymizer for Pulsar Logging](pip-402.md) |
+| PIP-404 | [Introduce per-ledger properties](pip-404.md) |
+| PIP-406 | [Introduce metrics related to dispatch throttled events](pip-406.md) |
+| PIP-407 | [Add a newMessage API to create a message with a schema and transaction](pip-407.md) |
+| PIP-409 | [support producer configuration for retry/dead letter topic producer](pip-409.md) |
+| PIP-412 | [Support setting messagePayloadProcessor in Pulsar Functions and Sinks](pip-412.md) |
+| PIP-414 | [Enforce topic consistency check](pip-414.md) |
+| PIP-415 | [Support getting message ID by index](pip-415.md) |
+| PIP-416 | [Add a new topic method to implement trigger offload by size threshold](pip-416.md) |
+| PIP-420 | [Provides an ability for Pulsar clients to integrate with third-party schema registry service](pip-420.md) |
+| PIP-421 | [Require Java 17 as the minimum for Pulsar Java client SDK](pip-421.md) |
+| PIP-422 | [Support global topic-level policy: replicated clusters and new API to delete topic-level policies](pip-422.md) |
+| PIP-425 | [Support connecting with next available endpoint for multi-endpoint serviceUrls](pip-425.md) |
+| PIP-427 | [Align pulsar-admin Default for Mark-Delete Rate with Broker Configuration](pip-427.md) |
+| PIP-428 | [Change TopicPoliciesService interface to fix consistency issues](pip-428.md) |
+| PIP-429 | [Optimize Handling of Compacted Last Entry by Skipping Payload Buffer Parsing](pip-429.md) |
+| PIP-430 | [Pulsar Broker cache improvements: refactoring eviction and adding a new cache strategy based on expected read count](pip-430.md) |
+| PIP-431 | [Add Creation and Last Publish Timestamps to Topic Stats](pip-431.md) |
+| PIP-432 | [Add isEncrypted field to EncryptionContext](pip-432.md) |
+| PIP-433 | [Optimize the conflicts of the replication and automatic creation mechanisms, including the automatic creation of topics and schemas](pip-433.md) |
+| PIP-435 | [Add `startTimestamp` and `endTimestamp` for consuming messages in client cli](pip-435.md) |
+| PIP-436 | [Add decryptFailListener to Consumer](pip-436.md) |
### Additional Information
diff --git a/pip/images/pip-148/image-001-a63d31.png b/pip/images/pip-148/image-001-a63d31.png
new file mode 100644
index 00000000000..a8814cc0449
Binary files /dev/null and b/pip/images/pip-148/image-001-a63d31.png differ
diff --git a/pip/images/pip-148/image-002-1ee1a6.png b/pip/images/pip-148/image-002-1ee1a6.png
new file mode 100644
index 00000000000..2e9bc4efbcf
Binary files /dev/null and b/pip/images/pip-148/image-002-1ee1a6.png differ
diff --git a/pip/images/pip-148/image-003-75b9f9.png b/pip/images/pip-148/image-003-75b9f9.png
new file mode 100644
index 00000000000..23788fa97b5
Binary files /dev/null and b/pip/images/pip-148/image-003-75b9f9.png differ
diff --git a/pip/images/pip-148/image-004-b09b05.png b/pip/images/pip-148/image-004-b09b05.png
new file mode 100644
index 00000000000..793e1448e7e
Binary files /dev/null and b/pip/images/pip-148/image-004-b09b05.png differ
diff --git a/pip/images/pip-148/image-005-bb0aee.png b/pip/images/pip-148/image-005-bb0aee.png
new file mode 100644
index 00000000000..009f7e27b3a
Binary files /dev/null and b/pip/images/pip-148/image-005-bb0aee.png differ
diff --git a/pip/images/pip-148/image-006-f51a36.png b/pip/images/pip-148/image-006-f51a36.png
new file mode 100644
index 00000000000..5915ab429a5
Binary files /dev/null and b/pip/images/pip-148/image-006-f51a36.png differ
diff --git a/pip/images/pip-148/image-007-a14ecc.png b/pip/images/pip-148/image-007-a14ecc.png
new file mode 100644
index 00000000000..2fae46399c2
Binary files /dev/null and b/pip/images/pip-148/image-007-a14ecc.png differ
diff --git a/pip/images/pip-31/image-001-17812c.png b/pip/images/pip-31/image-001-17812c.png
new file mode 100644
index 00000000000..29d141f779d
Binary files /dev/null and b/pip/images/pip-31/image-001-17812c.png differ
diff --git a/pip/images/pip-31/image-002-e0ba17.png b/pip/images/pip-31/image-002-e0ba17.png
new file mode 100644
index 00000000000..4c4550f6b21
Binary files /dev/null and b/pip/images/pip-31/image-002-e0ba17.png differ
diff --git a/pip/images/pip-31/image-003-3cc02b.png b/pip/images/pip-31/image-003-3cc02b.png
new file mode 100644
index 00000000000..ca62c733e80
Binary files /dev/null and b/pip/images/pip-31/image-003-3cc02b.png differ
diff --git a/pip/images/pip-31/image-004-ad2169.png b/pip/images/pip-31/image-004-ad2169.png
new file mode 100644
index 00000000000..a5225814628
Binary files /dev/null and b/pip/images/pip-31/image-004-ad2169.png differ
diff --git a/pip/images/pip-39/image-001-4633a3.png b/pip/images/pip-39/image-001-4633a3.png
new file mode 100644
index 00000000000..a77e3304379
Binary files /dev/null and b/pip/images/pip-39/image-001-4633a3.png differ
diff --git a/pip/images/pip-40/image-001-d0ea56.png b/pip/images/pip-40/image-001-d0ea56.png
new file mode 100644
index 00000000000..99c7b52a27e
Binary files /dev/null and b/pip/images/pip-40/image-001-d0ea56.png differ
diff --git a/pip/images/pip-40/image-002-6913c4.png b/pip/images/pip-40/image-002-6913c4.png
new file mode 100644
index 00000000000..109e90f83d2
Binary files /dev/null and b/pip/images/pip-40/image-002-6913c4.png differ
diff --git a/pip/images/pip-40/image-003-b570dd.png b/pip/images/pip-40/image-003-b570dd.png
new file mode 100644
index 00000000000..12beb50848f
Binary files /dev/null and b/pip/images/pip-40/image-003-b570dd.png differ
diff --git a/pip/images/pip-40/image-004-28764f.png b/pip/images/pip-40/image-004-28764f.png
new file mode 100644
index 00000000000..0842c25ed73
Binary files /dev/null and b/pip/images/pip-40/image-004-28764f.png differ
diff --git a/pip/images/pip-40/image-005-601b2f.png b/pip/images/pip-40/image-005-601b2f.png
new file mode 100644
index 00000000000..de17ac5ab6e
Binary files /dev/null and b/pip/images/pip-40/image-005-601b2f.png differ
diff --git a/pip/images/pip-40/image-006-d63ca4.png b/pip/images/pip-40/image-006-d63ca4.png
new file mode 100644
index 00000000000..40a005138b5
Binary files /dev/null and b/pip/images/pip-40/image-006-d63ca4.png differ
diff --git a/pip/images/pip-40/image-007-0eb68a.png b/pip/images/pip-40/image-007-0eb68a.png
new file mode 100644
index 00000000000..e4d16eecd93
Binary files /dev/null and b/pip/images/pip-40/image-007-0eb68a.png differ
diff --git a/pip/images/pip-40/image-008-29ff41.png b/pip/images/pip-40/image-008-29ff41.png
new file mode 100644
index 00000000000..8d31c9b2ec3
Binary files /dev/null and b/pip/images/pip-40/image-008-29ff41.png differ
diff --git a/pip/pip-148.md b/pip/pip-148.md
index e00c7991021..d3fc0ff5871 100644
--- a/pip/pip-148.md
+++ b/pip/pip-148.md
@@ -47,20 +47,20 @@ In the new Pulsar doc IA, the [release note belongs to the Reference chapter](ht
[https://docs.pingcap.com/tidb/stable/release-notes](https://docs.pingcap.com/tidb/stable/release-notes)
-![][image1]
+
* Add timeline table, including version NO., release date, download link, release naming conventions.
[https://docs.pingcap.com/tidb/stable/release-timeline](https://docs.pingcap.com/tidb/stable/release-timeline)
-![][image2]
+
[https://www.cockroachlabs.com/docs/releases/index.html](https://www.cockroachlabs.com/docs/releases/index.html)
-![][image3]
+
* On the release note page for each version, create the right navigation.
[https://docs.pingcap.com/tidb/stable/release-5.4.0](https://docs.pingcap.com/tidb/stable/release-5.4.0)
-![][image4]
+
* Make each release has its own .md file
@@ -70,13 +70,13 @@ In the new Pulsar doc IA, the [release note belongs to the Reference chapter](ht
### **Solution 1: Add version NO. to release note URL.**
-![][image5]
+
I’ve discussed with engineers:
* Many of them pay attention to URLs when scanning websites.
* Note: if we implement this enhancement, version NO. is only shown after clicking the link icon, or else (eg. scrolling up and down but not clicking anything) it is not shown
-![][image6]
+
### **Solution 2: add version NO. after headings**
@@ -91,7 +91,7 @@ We deprecated it since:
We deprecated it temporary since it is more difficult to implement (compared with solution 1\)
[https://docs.konghq.com/gateway/changelog/](https://docs.konghq.com/gateway/changelog/)
-![][image7]
+
## **Deprecated**
@@ -117,17 +117,3 @@ The following issues are to be resolved:
- [ ] On the Node.js release note page
- [ ] Change the format of “version.NO. \- release date” (headings) → running text
- [ ] Show heading 1 on the right navigation panel
-
-[image1]: <data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAnAAAAHQCAYAAAAh51fQAABh6klEQVR4XuzdZ5QUVdsv/OfL+XDWOee5z/ue8z5Lb/VWbzFgYCQnQQQREFAyKKCAkoyEIecMgkiUJFlyBsk55wwSBoY0wOScA9fb1252TXV1z1QPVdVTXfP/rbVXVe29u7qZ2dP7z+5Q/0EAAAAAEFT+Q1sBAAAAAPZWaIB78uQJCgoKCgoKCgpKMZTCeAQ47Q1RUFBQUFBQUFDsUdSUAKftpC15eXkoKCgoKCgoKCgWFm3+0haPAKdt9De05ebmoqCgoKCgoKCgPEPR5ipt0eYydYgrMMBpT6IObDk5OV4lOzsbBQUFBQUFBQWlkKLNT1wKC3TafFZogNPeWB3a5APIyspCQUFBQUFBQUExULTBzleQ0+Y0nwHOV3DzFdoyMzNFycjIQEFBQUFBQUFBKUKROaq [...]
-
-[image2]: <data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAnAAAAGDCAYAAACr/S2JAABnHUlEQVR4XuzddXAcV7c2+vPPrVsXvnNufVW3vuTF+74niWOUZGa2Y2ZmpsROzMyOEyeGmNmxHTPGzMzMLFtgi1mjEVh21p2127vV3TMjjZwZd8/4+VWt6t67d/eABh7tHmn+gwAAAADAr/yHsQMAAAAArC3fAPfHH3+gUCgUCoVCoUyo/OgCnHFHFAqFQqFQKJQ1SksNcMZBxnr79i0KhUKhUCgUyodlzF/G0gU448bChrY3b96gUCgUCoVCoTwoY45yV8Zcpg1xbgOc8SDaoJabm4tCoVAoFAqF8kIVFOyMGc1tgDPuqA1tr1+/VisnJweFQqFQKBQK9R6lzVTaMGfMYcac5jLAGYObDG/GwJadnY1CoVAoFAqF+hNlDHTuZuScAlx+4U0GNwAAAAD4MLRBzl2 [...]
-
-[image3]: <data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAnAAAAF4CAYAAADUqMRTAABpdUlEQVR4XuzdBXjcVro38Ha3d+/C3e29i7e7e9vtt23TNkxOHGZmZoYmaZgZGmoYG2ZmZmaHG45DDhnimGKM42TeT+85lkaSx45xQP7/nuc85+gcScOa/2hmpA8IAAAAADzKB+YOAAAAAHBvWoCz2WwoKCgoKCgoKChuXAwBzjyolnfv3qGgoKCgoKCgoLigmHOZPsQ5DHDmFXB5+/YtCgoKCgoKCgpKFhZz/nIU5ESASy648UoSEhJEefPmDcXHx6OgoKCgoKCgoGRh4cyl5i9zoNNnNocBTg1v+uD2+vVriouLEyU2NhYFBQUFBQUFBSUTipqvOGuZg5w+xDkMcPrwpgY3AAAAAHANNcg5CnFJAhynPXWPGwAAAAC4hrpHjrNZigFO3fvGC/AuPQAAAABwDc5 [...]
-
-[image4]: <data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAnAAAAGsCAYAAABdB5pqAACAAElEQVR4Xuyd9Vscy9ao7z9wf79nH9nfOd+2KJKgIQTixN3d3d3d3d1dibu7E08gQAIEdwiQhJx1e62aqulpBgIJQzLDep+nnpKutpme6Xeqe3r9H2AYhmEYhmHsiv9jbGAYhmEYhmF+bqwK3H//+19OnDhx4sSJEydOP0GyhhI4Y2dj+vLlCydOnDhx4sSJEycbJqN/GVOhBM64UE6cOHHixIkTJ04lk4xelkfgjBONC8jNzVXp8+fPnDhx4sSJEydOnIox6V3L6GFGT7MqcNbETS7806dP8PHjR06cOHHixIkTJ07FmNCxMOllLj+JyyNwRnkzSltOTg5kZ2dTysrK4sSJEydOnDhx4vQdSXoVOpZe5qTIWZM4Ejhro296edPPrB/e48SJEydOnDhx4lR8Se9 [...]
-
-[image5]: <data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAnAAAAE4CAYAAADb8M4/AACAAElEQVR4XuydBZhWxdvG1/rbn510SzcigggGoICIKGU3iqKAIK2IYoItFiomYmEhYdCIdAhI5y5LSTfPN/fMPHPmzDnv7rvLCijzXNd9TZ458Z74nWfmzJtC3rx58+bNmzdv3v5VluJmePPmzZs3b968eTu0LcsAt2/fPi8vLy8vLy8vrxxUVi0CcG6DXl5eXl5eXl5eB1euhQDOrWxr7969sdqzZ4+Xl5eXl5eXl1cW5PIUy+WvRBBnAM6tlAjSdu/eHdKuXbu8vLy8vLy8vLyyIJenEsGdy2chgHML4+DNBbadO3eGtGPHDi8vLy8vLy8vrwzk8pMLdMlCXATg4sDNhjasfPv27Ubbtm3z8vLy8vLy8vLKgmyWYrCzYc4FuaQAzvW42eCGlW7dupW2bNlCmzd [...]
-
-[image6]: <data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAnAAAAFbCAYAAABVkLPLAACAAElEQVR4XuydB5gVRdb3effbffcVJYmkYZgZsqASFUREBQUURTHnhAkVxIw5RwyY0BUVEIwrggqIAioiihkQFFGyBEnKoq6Spr4+VXWqT5/uvrfvzJ1hZjj/5/lTVaeq+vbtG+bHqe6+lZRIJBKJRCKRqFypEg+IRCKRSCQSicq2BOBEIpFIJBKJypkE4EQikUgkEonKmRIBXGFhoVgsFovFYrG4FJxEkQDHN5TK27dvF4vFYrFYLBZnYM5TqRylEMDxSdx8B6i3bdsmFovFYrFYLI4w56ZMgI4rAHB8cCpo4zsF3rp1q1gsFovFYrE4wpyb4qCOc1gUxDmA44Pi4C0O2LZs2RLy5s2bxWKxWCwWi3dKcy4CxwEd5y3OYxziYgGObygVtNGd/euvv0L+888/xWK [...]
-
-[image7]: <data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAnAAAAGDCAYAAACr/S2JAACAAElEQVR4XuydBZgUx7bHc9+7Sa7k5l2L3ZuEECwhAYKGICEQHIIT3N0hwd3dLUhwCxI0uLv7YrvLLrAu+C62C+f1Od1VW109swKzMEvO//v+X1WdqpbpafnN6emZV4DFYrFYLBaLlab0ih5gsVgsFovFYnm3JMA9efLEpR8/fsxms9lsNpvNfo7WeUzYBnBJAVt8fDybzWaz2Ww2+zlY5zAd6BwAp0NbXFwcPHr0iPzw4UM2m81ms9lsdipacBcymA5zNoDT4Q0HC2B78OAB3L9/H+7duwexsbFsNpvNZrPZ7FQ0MheyFzKYADoV5AS32QBOhbfrq08ZPgnX15xSfFKJYWnZNk7EtWkc80rEa0+79rozEE0+DdG/nTG93oU3nDWMpaibjlLq0Rt8jLaPWW40vMm [...]
\ No newline at end of file
diff --git a/pip/pip-31.md b/pip/pip-31.md
index 559bdb5ab2a..9cd09aef750 100644
--- a/pip/pip-31.md
+++ b/pip/pip-31.md
@@ -1,4 +1,4 @@
-# PIP-31: Transaction Support
+# PIP-31: Transaction Support (Transactional Streaming)
* **Status**: Implemented
* **Author**: Sijie Guo
@@ -6,253 +6,39 @@
* **Mailing List discussion**:
* **Release**: 2.8.0
-[**Motivation**](#motivation) **[4](#motivation)**
-
-[Use Cases](#use-cases) [4](#use-cases)
-
-[Event Processing](#event-processing) [4](#event-processing)
-
-[Atomic Produce](#atomic-produce) [5](#atomic-produce)
-
-[Transactional Guarantees](#transactional-guarantees) [**6**](#transactional-guarantees)
-
-[Isolation Levels](#isolation-levels) [6](#isolation-levels)
-
-[A 10,000 Foot View](#a-10,000-foot-view) [**7**](#a-10,000-foot-view)
-
-[Concepts](#concepts) [7](#concepts)
-
-[Transaction Coordinator](#transaction-coordinator) [7](#transaction-coordinator)
-
-[Transaction Buffer](#transaction-buffer) [7](#transaction-buffer)
-
-[Acknowledgements in Transactions](#acknowledgements-in-transactions) [8](#acknowledgements-in-transactions)
-
-[Materialization Mechanism](#materialization-mechanism) [8](#materialization-mechanism)
-
-[Transaction Flow](#transaction-flow) [9](#transaction-flow)
-
-[1\. Begin Transaction](#1.-begin-transaction) [10](#1.-begin-transaction)
-
-[2\. Transaction Loop](#2.-transaction-loop) [10](#2.-transaction-loop)
-
-[2.1 Add Partitions To Txn](#2.1-add-partitions-to-txn) [10](#2.1-add-partitions-to-txn)
-
-[2.2 Produce Messages to Partitions](#2.2-produce-messages-to-partitions) [10](#2.2-produce-messages-to-partitions)
-
-[2.3 Add Subscriptions To Txn](#2.3-add-subscriptions-to-txn) [11](#2.3-add-subscriptions-to-txn)
-
-[2.4 Ack Messages on Subscriptions](#2.4-ack-messages-on-subscriptions) [11](#2.4-ack-messages-on-subscriptions)
-
-[3\. End Transaction](#3.-end-transaction) [11](#3.-end-transaction)
-
-[3.1 End Transaction Request](#3.1-end-transaction-request) [11](#3.1-end-transaction-request)
-
-[3.2 Finalizing Process](#3.2-finalizing-process) [12](#3.2-finalizing-process)
-
-[3.3 Mark Transaction as COMMITTED or ABORTED](#3.3-mark-transaction-as-committed-or-aborted) [12](#3.3-mark-transaction-as-committed-or-aborted)
-
-[**Design Choices**](#design-choices) **[13](#design-choices)**
-
-[Transaction Buffer](#transaction-buffer-1) [13](#transaction-buffer-1)
-
-[Marker Approach](#marker-approach) [13](#marker-approach)
-
-[Overview](#overview) [14](#overview)
-
-[Challenges](#challenges) [15](#challenges)
-
-[Materialization](#materialization) [15](#materialization)
-
-[Sweep](#sweep) [15](#sweep)
-
-[Retention](#retention) [16](#retention)
-
-[Changes](#changes) [16](#changes)
-
-[Discussions](#discussions) [16](#discussions)
-
-[Sidecar Approach](#sidecar-approach) [17](#sidecar-approach)
-
-[Overview](#overview-1) [17](#overview-1)
-
-[Challenges](#challenges-1) [19](#challenges-1)
-
-[Materialization](#materialization-1) [19](#materialization-1)
-
-[Sweep](#sweep-1) [20](#sweep-1)
-
-[Retention](#retention-1) [20](#retention-1)
-
-[Changes](#changes-1) [20](#changes-1)
-
-[Discussions](#discussions-1) [20](#discussions-1)
-
-[Huge Size Transaction](#huge-size-transaction) [20](#huge-size-transaction)
-
-[**A Full Proposal**](#a-full-proposal) **[21](#a-full-proposal)**
-
-[Public APIs](#public-apis) [21](#public-apis)
-
-[Examples](#examples) [24](#examples)
-
-[Data Structures](#data-structures) [25](#data-structures)
-
-[TxnID](#txnid) [25](#txnid)
-
-[Wire Protocol Changes](#wire-protocol-changes) [25](#wire-protocol-changes)
-
-[New Commands](#new-commands) [25](#new-commands)
-
-[TxnAction](#txnaction) [25](#txnaction)
-
-[CommandNewTxn](#commandnewtxn) [25](#commandnewtxn)
-
-[CommandNewTxnResponse](#commandnewtxnresponse) [26](#commandnewtxnresponse)
-
-[CommandAddPartitionToTxn](#commandaddpartitiontotxn) [26](#commandaddpartitiontotxn)
-
-[CommandAddPartitionToTxnResponse](#commandaddpartitiontotxnresponse) [26](#commandaddpartitiontotxnresponse)
-
-[CommandAddSubscritpionToTxn](#commandaddsubscritpiontotxn) [27](#commandaddsubscritpiontotxn)
-
-[CommandAddSubscriptionToTxnResponse](#commandaddsubscriptiontotxnresponse) [27](#commandaddsubscriptiontotxnresponse)
-
-[CommandEndTxn](#commandendtxn) [28](#commandendtxn)
-
-[CommandEndTxnResponse](#commandendtxnresponse) [28](#commandendtxnresponse)
-
-[CommandEndTxnOnPartition](#commandendtxnonpartition) [28](#commandendtxnonpartition)
-
-[CommandEndTxnOnPartitionResponse](#commandendtxnonpartitionresponse) [29](#commandendtxnonpartitionresponse)
-
-[CommandEndTxnOnSubscription](#commandendtxnonsubscription) [29](#commandendtxnonsubscription)
-
-[CommandEndTxnOnSubscriptionResponse](#commandendtxnonsubscriptionresponse) [29](#commandendtxnonsubscriptionresponse)
-
-[Modified Commands](#modified-commands) [30](#modified-commands)
-
-[CommandSend](#commandsend) [30](#commandsend)
-
-[CommandAck](#commandack) [30](#commandack)
-
-[Transaction Coordinator](#transaction-coordinator-1) [31](#transaction-coordinator-1)
-
-[Concepts](#concepts-1) [33](#concepts-1)
-
-[TCID](#tcid) [33](#tcid)
-
-[TxnStatus](#txnstatus) [33](#txnstatus)
-
-[TransactionMetadataStoreProvider](#transactionmetadatastoreprovider) [33](#transactionmetadatastoreprovider)
-
-[TransactionMetadataStore](#transactionmetadatastore) [34](#transactionmetadatastore)
-
-[Implementation](#implementation) [34](#implementation)
-
-[Transaction Log](#transaction-log) [34](#transaction-log)
-
-[TC Startup](#tc-startup) [35](#tc-startup)
-
-[TC Request Processing](#tc-request-processing) [36](#tc-request-processing)
-
-[CommandNewTxn](#commandnewtxn-1) [36](#commandnewtxn-1)
-
-[CommandAddPartitionToTxn](#commandaddpartitiontotxn-1) [36](#commandaddpartitiontotxn-1)
-
-[CommandAddSubscriptionToTxn](#commandaddsubscriptiontotxn) [37](#commandaddsubscriptiontotxn)
-
-[CommandEndTxn](#commandendtxn-1) [37](#commandendtxn-1)
-
-[TC Transaction Expiration](#tc-transaction-expiration) [38](#tc-transaction-expiration)
-
-[Deployment](#deployment) [38](#deployment)
-
-[TC in Broker](#tc-in-broker) [38](#tc-in-broker)
-
-[Broker \- Transaction Buffer](#broker---transaction-buffer) [39](#broker---transaction-buffer)
-
-[Message](#message) [39](#message)
-
-[Message ID](#message-id) [39](#message-id)
-
-[Transaction Log](#transaction-log-1) [40](#transaction-log-1)
-
-[Implementation](#implementation-1) [40](#implementation-1)
-
-[Transaction Manager](#transaction-manager) [40](#transaction-manager)
-
-[Dedup Cursor](#dedup-cursor) [40](#dedup-cursor)
-
-[Transaction Cursor](#transaction-cursor) [40](#transaction-cursor)
-
-[Transaction Index](#transaction-index) [40](#transaction-index)
-
-[Retention Cursor](#retention-cursor) [41](#retention-cursor)
-
-[Compacting Transaction Log](#compacting-transaction-log) [41](#compacting-transaction-log)
-
-[Partition Loading](#partition-loading) [42](#partition-loading)
-
-[Broker Request Processing](#broker-request-processing) [42](#broker-request-processing)
-
-[CommandSend](#commandsend-1) [42](#commandsend-1)
-
-[CommandEndTxnOnPartition](#commandendtxnonpartition-1) [42](#commandendtxnonpartition-1)
-
-[Broker Dispatching](#broker-dispatching) [43](#broker-dispatching)
-
-[Broker \- Acknowledgements](#broker---acknowledgements) [43](#broker---acknowledgements)
-
-[New Acknowledgement State](#new-acknowledgement-state) [44](#new-acknowledgement-state)
-
-[PENDING\_ACK State](#pending_ack-state) [45](#pending_ack-state)
-
-[Negative Acknowledgement](#negative-acknowledgement) [46](#negative-acknowledgement)
-
-[Client-Broker Request Processing](#client-broker-request-processing) [46](#client-broker-request-processing)
-
-[**Future Worker**](#future-worker) **[46](#future-worker)**
-
-[Single Partition Transaction](#single-partition-transaction) [46](#single-partition-transaction)
-
-#
-
-# PIP-31: Transactional Streaming
-
-# Motivation {#motivation}
+# Motivation <a id="motivation"></a>
This document outlines a proposal for supporting transactional messaging at Apache Pulsar. Transactions are used for strengthening the message delivery semantics of Apache Pulsar and processing guarantees at Pulsar Functions.
-The highest message delivery guarantee that Apache Pulsar currently provides is \`exactly-once\` producing at one single partition via Idempotent Producer. Users are guaranteed that every message produced to one single partition via an Idempotent Producer will be persisted exactly once, without data loss. There is no \`atomicity\` when a producer attempts to produce messages to multiple partitions. For instance, a publish failure can occur when the broker crashes and if the producer does [...]
+The highest message delivery guarantee that Apache Pulsar currently provides is `exactly-once` producing at one single partition via Idempotent Producer. Users are guaranteed that every message produced to one single partition via an Idempotent Producer will be persisted exactly once, without data loss. There is no `atomicity` when a producer attempts to produce messages to multiple partitions. For instance, a publish failure can occur when the broker crashes and if the producer doesn’t [...]
Similarly, Pulsar Functions only guarantees exactly-once processing on a single event on idempotent function. It can’t guarantee processing multiple events or producing multiple results can happen exactly once. For example, if a function accepts multiple events and produces one result (e.g. windowing functions), the function can fail between producing the result and acknowledging the incoming messages, or even between acknowledging individual events. This will cause all (or some) incomin [...]
-Users of Pulsar and Pulsar Functions will greatly benefit from transactional semantic support. Every message written or processed will be happening exactly once, without duplicates and without data loss \- even in the event of broker or function instance failures. A transactional messaging semantic not only makes writing applications using Pulsar or Pulsar Functions easier, it expands the scope which Pulsar can provide.
+Users of Pulsar and Pulsar Functions will greatly benefit from transactional semantic support. Every message written or processed will be happening exactly once, without duplicates and without data loss - even in the event of broker or function instance failures. A transactional messaging semantic not only makes writing applications using Pulsar or Pulsar Functions easier, it expands the scope which Pulsar can provide.
-## Use Cases {#use-cases}
+## Use Cases <a id="use-cases"></a>
-### Event Processing {#event-processing}
+### Event Processing <a id="event-processing"></a>
Event Processing (aka Event Streaming, Stream Processing) applications will hugely benefit from transactional guarantees. Event processing application, which is typically a pipeline of ‘consume-transform-produce’ tasks, requires transactional guarantees when duplicate processing of the event stream is unacceptable.
-Pulsar Functions is one of such event processing framework built for Apache Pulsar. A function invocation is a typical \`consume-transform-produce\` task \- It reads events from one (or multiple) topic(s), processes the events by invoking the function, and produces the results to one (or multiple) topic(s). [^1][^2]
+Pulsar Functions is one of such event processing framework built for Apache Pulsar. A function invocation is a typical `consume-transform-produce` task - It reads events from one (or multiple) topic(s), processes the events by invoking the function, and produces the results to one (or multiple) topic(s). [^1][^2]
-Many SPE (stream processing engines) or customized event processing logic (using Pulsar producer and consumer) falls into the category of \`consume-transform-produce\`. It requires the ability to produce a batch of messages to different topic-partitions and acknowledge a batch of messages in one atomic operation. By atomically, we mean either all messages are committed (all output messages are persisted exactly-once and all input messages are acknowledged), or none of them are.
+Many SPE (stream processing engines) or customized event processing logic (using Pulsar producer and consumer) falls into the category of `consume-transform-produce`. It requires the ability to produce a batch of messages to different topic-partitions and acknowledge a batch of messages in one atomic operation. By atomically, we mean either all messages are committed (all output messages are persisted exactly-once and all input messages are acknowledged), or none of them are.
-The number of messages produced and number of messages acknowledged in a single transaction can vary \- from a few messages or a ton of messages. The number of topic-partitions (aka partitions)[^3] can vary \- from one single partition to many partitions. It depends on the processing efficiency, windowing/buffering logic and many other application related factors. A typical example for understanding the variants described above is windowing operators in an event processing application. A [...]
+The number of messages produced and number of messages acknowledged in a single transaction can vary - from a few messages or a ton of messages. The number of topic-partitions (aka partitions)[^3] can vary - from one single partition to many partitions. It depends on the processing efficiency, windowing/buffering logic and many other application related factors. A typical example for understanding the variants described above is windowing operators in an event processing application. As [...]
-### Atomic Produce {#atomic-produce}
+### Atomic Produce <a id="atomic-produce"></a>
-Atomically producing multiple messages is a variant of \`consume-transfer-produce\`, where it only requires \`produce\` messages in an atomic way.
+Atomically producing multiple messages is a variant of `consume-transfer-produce`, where it only requires `produce` messages in an atomic way.
Another variant of atomic-produce is for going beyond the 5MB message size limitation in Apache Pulsar. You can increase the limit of message to a very large message size, but it means you will have very large network and disk IO, which is not friendly to memory and disk utilization. People will end up breaking down a large message into multiple smaller chunks, and send each chunk as one pulsar message. If doing so, we need to make sure producing such chunks are done in an atomic way.
Database CDC or using Apache Pulsar as commit log are typical use cases for atomic producing.
-# Transactional Guarantees {#transactional-guarantees}
+# Transactional Guarantees <a id="transactional-guarantees"></a>
-As described in [Motivation](#bookmark=id.g2xdepks9p63) section, providing transactional semantics will enable event streaming applications to consume, process, and produce messages in one atomic operation.
+As described in [Motivation](#motivation) section, providing transactional semantics will enable event streaming applications to consume, process, and produce messages in one atomic operation.
That means, ***a batch of messages in a transaction can be received from, produced to and acknowledged to many partitions. All the operations involved in a transaction will succeed or fail as one single unit.***
@@ -261,31 +47,31 @@ However, we are not guaranteed the messages produced within a committed transact
1. Consumers may not consume from all the partitions which participated in the committed transaction. Hence they will never be able to read all the messages that comprised in the transaction.
2. Consumers may have a different receiver queue size or buffering/window size, which they might only be interested in consuming a certain amount of messages. That amount can be any arbitrary number.
-*However, we might be able to support consuming messages committed to one single partition all together. But it depends on the design we choose below. As there is no strong requirement for this feature, I will leave this point outside of guarantees for now.*
+However, we might be able to support consuming messages committed to one single partition all together. But it depends on the design we choose below. As there is no strong requirement for this feature, I will leave this point outside of guarantees for now.
-## Isolation Levels {#isolation-levels}
+## Isolation Levels <a id="isolation-levels"></a>
Similar as Database Transactions, Transactions in an event streaming system will also have isolation levels. The isolations levels are:
-- READ\_UNCOMMITTED: It means no isolation and consumers are able to read uncommitted messages.
-- READ\_COMMITTED: All consumers are only able to read committed messages.
+- READ_UNCOMMITTED: It means no isolation and consumers are able to read uncommitted messages.
+- READ_COMMITTED: All consumers are only able to read committed messages.
- SERIALIZABILITY: The execution of multiple transactions over multiple partitions is equivalent to some serial execution (total ordering) of the transactions. This guarantees the committed messages of transaction A appear before the committed messages of transaction B if transaction A is committed before B. Hence, the consumer will be able to see exactly the same order of the committed messages across the partitions involved in those transactions.
-The isolation level that Pulsar transaction must support is **READ\_COMMITTED**. Whether to support READ\_UNCOMMITTED or SERIALIZABILITY requires inputs from Pulsar users.
+The isolation level that Pulsar transaction must support is **READ_COMMITTED**. Whether to support READ_UNCOMMITTED or SERIALIZABILITY requires inputs from Pulsar users.
-# A 10,000 Foot View {#a-10,000-foot-view}
+# A 10,000 Foot View <a id="a-10,000-foot-view"></a>
There are many ways to design the transaction support in Pulsar. All these proposals can be shaped into one common framework, which we will discuss in this section. Further sections will describe the details based on this framework in detail.
-## Concepts {#concepts}
+## Concepts <a id="concepts"></a>
-### Transaction Coordinator {#transaction-coordinator}
+### Transaction Coordinator <a id="transaction-coordinator"></a>
In order to achieve transactional messaging, we will have to introduce a server-side module named ***Transaction Coordinator*** (aka TC). TC manages transactions of messages sent by producers and acknowledgements sent by consumers, and commit or abort the operations as a whole.
The Transaction Coordinator will be persisting transaction statuses in some persistent storage (e.g. a transaction log backed by a separated topic, or a table in table service) for recovery. We will talk about how a TC can be implemented and how it manages the transaction statuses in further section.
-### Transaction Buffer {#transaction-buffer}
+### Transaction Buffer <a id="transaction-buffer"></a>
Messages produced within a transaction will be stored in ***Transaction Buffer*** (aka TB). The messages in TB will not be materialized (visible) to consumers until their transactions are committed. The messages in TB will be discarded when their transactions are aborted. A cleanup process (e.g. compaction) might be required to clean up messages of aborted transactions depending on how TB is implemented.
@@ -296,30 +82,31 @@ A TB implementation requires :
There are other factors to be considered such as cleaning up messages of aborted transactions, write amplifications, ordering, and etc. We will discuss the solutions of how a TB can be implemented and what are the tradeoffs.
-### Acknowledgements in Transactions {#acknowledgements-in-transactions}
+### Acknowledgements in Transactions <a id="acknowledgements-in-transactions"></a>
-Many event streaming applications (e.g. Pulsar Functions) include both consumers and producers, where the applications consume messages from input Pulsar topics and produce new messages to output Pulsar topics. To achieve \`exactly once\` streaming, we need to make sure acknowledgements on the input messages happen as part of the transaction in order to achieve atomicity. Otherwise, if there is a failure between acknowledging inputs and producing the messages to output topics, data dupli [...]
+Many event streaming applications (e.g. Pulsar Functions) include both consumers and producers, where the applications consume messages from input Pulsar topics and produce new messages to output Pulsar topics. To achieve `exactly once` streaming, we need to make sure acknowledgements on the input messages happen as part of the transaction in order to achieve atomicity. Otherwise, if there is a failure between acknowledging inputs and producing the messages to output topics, data duplica [...]
Hence, we need to include acknowledgments in transactions to guarantee atomicity. In order to achieve this, we have to change the behavior of acknowledgements in a transaction. Because currently all the acknowledgements in Pulsar are only best-effort operations, ack’s can be lost during network disconnections or broker crashes, which would result in data duplicates.
We also need to consider commit conflicts between individual acknowledgements and cumulative acknowledgements. We will talk about how to enhance consumer protocol and cursor management for supporting acknowledgements in transactions in further sections.
-### Materialization Mechanism {#materialization-mechanism}
+### Materialization Mechanism <a id="materialization-mechanism"></a>
For messages appended to TB, the transaction implementation should also provide materialize mechanism to materialize uncommitted messages to make them visible to consumer when transactions are committed. This materialize mechanism varies based on the implementation of TB.
-The materialization mechanism should also be considering isolation levels (if we want to support higher isolation level than READ\_COMMITTED).
+The materialization mechanism should also be considering isolation levels (if we want to support higher isolation level than READ_COMMITTED).
We will talk about how transaction materializes uncommitted messages in further sections.
-## Transaction Flow {#transaction-flow}
+## Transaction Flow <a id="transaction-flow"></a>
All transaction implementations can be shaped using these key components/concepts described in the above sections.
-![][image1]
-Figure 1\. Transaction Flow
+
+
+Figure 1. Transaction Flow
-In [Figure 1\. Transaction Flow](#bookmark=id.5x2yklb5p0ig):
+In the above figure:
- The gray square boxes represent distinct brokers.
- The gray rounded boxes represent logical components which can be running inside a broker or as a separated service (e.g. like how we run function worker as part of broker)
@@ -327,19 +114,19 @@ In [Figure 1\. Transaction Flow](#bookmark=id.5x2yklb5p0ig):
- Each arrow represents the request flow or message flow. These operations occur in sequence indicated by the numbers next to each arrow.
- The sections below are numbered to match the operations showed in the diagram.
-### 1\. Begin Transaction {#1.-begin-transaction}
+### 1. Begin Transaction <a id="1.-begin-transaction"></a>
At the beginning of a transaction, the pulsar client will find a Transaction Coordinator. The TC will allocate a transaction id (aka TxnID) for the transaction. The transaction will be logged with its transaction id and a status of OPEN (indicating the transaction is OPEN) in the transaction log (as shown in step 1a). This ensures the transaction status is persisted regardless TC crashes. After a transaction status entry is logged, TC responds the transaction id back to the pulsar client.
-### 2\. Transaction Loop {#2.-transaction-loop}
+### 2. Transaction Loop <a id="2.-transaction-loop"></a>
In this stage, the pulsar client will enter a transaction loop, repeating the actions of consume-transform-produce the messages that comprise the transaction. This is a long phase and is potentially comprised of multiple produce and acknowledgement requests.
-#### 2.1 Add Partitions To Txn {#2.1-add-partitions-to-txn}
+#### 2.1 Add Partitions To Txn <a id="2.1-add-partitions-to-txn"></a>
Before the pulsar client produces messages to a new topic partition, the client sends a request to TC to add the partition to the transaction. TC logs the partitions changes of the transaction into its transaction log for durability (as shown in 2.1a). This step ensures TC knows all the partitions that a transaction is touching, so TC can commit or abort changes on each partition at the end-partition phase.
-#### 2.2 Produce Messages to Partitions {#2.2-produce-messages-to-partitions}
+#### 2.2 Produce Messages to Partitions <a id="2.2-produce-messages-to-partitions"></a>
The pulsar client starts producing messages to partitions. This produce flow is same as the normal message produce flow. The only difference is the batch of messages produced by a transaction will be containing the transaction id. The broker receiving the batch of messages checks if the batch of messages belongs to a transaction. If it doesn’t belong to a transaction, the broker writes the batch directly into the partition’s managed ledger (it is the normal produce flow). If it belongs t [...]
@@ -351,21 +138,21 @@ The transaction buffer must meet the following requirements:
The transaction buffer can be implemented in many ways. It can be managed ledger itself, a separated sidecar managed ledger, or some other implementation. We will be discussing more details about the design choices of transaction buffer in further sections.
-#### 2.3 Add Subscriptions To Txn {#2.3-add-subscriptions-to-txn}
+#### 2.3 Add Subscriptions To Txn <a id="2.3-add-subscriptions-to-txn"></a>
The pulsar client sends a request to TC the first time a new subscription is acknowledged to as part of a transaction. The addition of the subscription to the transaction is logged by TC in step 2.3a. This step ensures TC knows all the subscription that a transaction is touching, so TC can commit or abort changes on each subscription at the EndTxn phase.
-#### 2.4 Ack Messages on Subscriptions {#2.4-ack-messages-on-subscriptions}
+#### 2.4 Ack Messages on Subscriptions <a id="2.4-ack-messages-on-subscriptions"></a>
-The pulsar client start acknowledging messages on subscriptions. This transactional acknowledgement flow is same as the normal acknowledgement flow. However the ack request will be carrying a transaction id. The broker receiving the acknowledgement request checks if the acknowledge belongs to a transaction or not. If it belongs to a transaction, the broker will mark the message as in PENDING\_ACK state. PENDING\_ACK state means the message can not be acknowledged or negative-acknowledged [...]
+The pulsar client start acknowledging messages on subscriptions. This transactional acknowledgement flow is same as the normal acknowledgement flow. However the ack request will be carrying a transaction id. The broker receiving the acknowledgement request checks if the acknowledge belongs to a transaction or not. If it belongs to a transaction, the broker will mark the message as in PENDING_ACK state. PENDING_ACK state means the message can not be acknowledged or negative-acknowledged b [...]
The pulsar client will abort the whole transaction when it tries to acknowledge but a conflict is detected. The conflict can be detected on both individual acks and cumulative acks.
-### 3\. End Transaction {#3.-end-transaction}
+### 3. End Transaction <a id="3.-end-transaction"></a>
At the end of a transaction, the application will decide to commit or abort the transaction. The transaction can also be aborted when a conflict is detected on acknowledging messages.
-#### 3.1 End Transaction Request {#3.1-end-transaction-request}
+#### 3.1 End Transaction Request <a id="3.1-end-transaction-request"></a>
When a pulsar client is finished with a transaction, it can issue an end transaction request to TC, with a field indicating whether the transaction is to be committed or aborted.
@@ -375,7 +162,7 @@ Upon receiving this request, the TC:
2. Begins the process of committing or aborting messages or acknowledgments to all the partitions involved in this transaction. It is shown in 3.2 and described in section 3.2 below.
3. After all the partitions involved in this transaction have been successfully committed or aborted, TC writes COMMITTED or ABORTED messages to its transaction log. It is shown in 3.3 in the diagram and described in section 3.3 below.
-#### 3.2 Finalizing Process {#3.2-finalizing-process}
+#### 3.2 Finalizing Process <a id="3.2-finalizing-process"></a>
At this stage, TC will finalize the transaction by committing or aborting messages or acknowledgments on all the partitions involved in this transaction.
@@ -383,9 +170,9 @@ Committing produced message is materializing the messages and making them visibl
Aborting produced message is discarding the messages in TB. TB has to ensure cleaning up those messages and reclaim spaces if the transaction is aborted.
-Committing acknowledgments moves a message from PENDING\_ACK to ACK. Aborting acknowledgments will negative acks a message, so the message will be re-deliver to other consumers.
+Committing acknowledgments moves a message from PENDING_ACK to ACK. Aborting acknowledgments will negative acks a message, so the message will be re-deliver to other consumers.
-#### 3.3 Mark Transaction as COMMITTED or ABORTED {#3.3-mark-transaction-as-committed-or-aborted}
+#### 3.3 Mark Transaction as COMMITTED or ABORTED <a id="3.3-mark-transaction-as-committed-or-aborted"></a>
After all produced messages and acknowledgements to all partitions are committed or aborted, the TC writes the final COMMITTED or ABORTED transaction status message to its transaction log, indicating that the transaction is complete (shown as 3.3a in the diagram). At this point, all the messages pertaining to the transaction in its transaction log can safely be removed.
@@ -395,11 +182,11 @@ This diagram shows the whole transaction flows involving different components. H
Also there are many optimizations can be made on improving the transaction flow. Those are left out of this proposal to make sure we start with a reliable and robust implementation to get things right first.
-# Design Choices {#design-choices}
+# Design Choices <a id="design-choices"></a>
-\`[Transaction Coordinator](#bookmark=id.s49iko5e8j8a)\` and \`[Transactional Acknowledgement](#bookmark=id.asf2ci9pcdeb)\` are pretty straightforward to implement. See details at Section “[A Full Proposal](#bookmark=id.qw636l7pty05)”. The most challenge part will be the \`Transaction Buffer\` part, as there will be many proposals with different tradeoffs there. These proposals are discussed below.
+[Transaction Coordinator](#transaction-coordinator) and [Transactional Acknowledgement](#broker---acknowledgements) are pretty straightforward to implement. See details at Section “[A Full Proposal](#a-full-proposal)”. The most challenge part will be the [Transaction Buffer](#transaction-buffer) part, as there will be many proposals with different tradeoffs there. These proposals are discussed below.
-## Transaction Buffer {#transaction-buffer-1}
+## Transaction Buffer <a id="transaction-buffer-1"></a>
Recapping what we have described of a transactional flow above, a transaction buffer implementation should consider followings:
@@ -411,7 +198,7 @@ Recapping what we have described of a transactional flow above, a transaction bu
1. How messages are materialized will impact how we dispatch messages to consumers
3. A sweep mechanism to discard the messages from transaction buffer for reclaiming disk spaces.
-### Marker Approach {#marker-approach}
+### Marker Approach <a id="marker-approach"></a>
One of the approaches to implement the transaction buffer is:
@@ -421,24 +208,25 @@ One of the approaches to implement the transaction buffer is:
- Mark the messages as visible when the transaction is committed
- Cleanup (in background) the messages when the transaction is aborted.
-#### Overview {#overview}
+#### Overview <a id="overview"></a>
-![][image2]
-Figure 2\. Marker Approach
+
-Figure 2 [Marker Approach](#bookmark=id.kc3lb5od0iy6) demonstrates how a marker approach looks like. The gray boxes represents the messages produced by a normal client (through non-transactional flow); the color boxes represents the messages produced by transactions; different color indicates different transaction. Each message produced by a transaction will be labelled as \`\<txn\>-\<msg\>\` (e.g. \`txn2-m2\`). \`\<txn\>-commit\` or \`\<txn\>-abort\` are the markers appended when commit [...]
+Figure 2. Marker Approach
-In this approach, all the transactional messages are appended directly to the partition’s managed ledger. Additional metadata (e.g. a TxnID field) is required to add to the messages when they are produced to brokers. The broker dispatcher checks those metadata and also the transaction status to decide whether it should dispatch them. Each transaction will be using TxnID as the producer id, so broker can use the de-duplication logic to ensure messages are appended to the partition exactly [...]
+The above figure demonstrates how a marker approach looks like. The gray boxes represents the messages produced by a normal client (through non-transactional flow); the color boxes represents the messages produced by transactions; different color indicates different transaction. Each message produced by a transaction will be labelled as `<txn>-<msg>` (e.g. `txn2-m2`). `<txn>-commit` or `<txn>-abort` are the markers appended when committing or aborting a given transaction.
-Figure 2 demonstrates 3 transactions, \`txn1\`, \`txn2\`, and \`txn3\`. \`txn1\` and \`txn2\` are committed, while \`txn3\` is aborted.
+In this approach, all the transactional messages are appended directly to the partition’s managed ledger. Additional metadata (e.g. a TxnID field) is required to add to the messages when they are produced to brokers. The broker dispatcher checks those metadata and also the transaction status to decide whether it should dispatch them. Each transaction will be using TxnID as the producer id, so broker can use the de-duplication logic to ensure messages are appended to the partition exactly [...]
-#### Challenges {#challenges}
+Figure 2 demonstrates 3 transactions, `txn1`, `txn2`, and `txn3`. `txn1` and `txn2` are committed, while `txn3` is aborted.
+
+#### Challenges <a id="challenges"></a>
There are a few challenges in this approach.
-##### Materialization {#materialization}
+##### Materialization <a id="materialization"></a>
-\`\<txn\>-commit\` is the commit marker used for marking the transaction as \`COMMITTED\` and materializing the messages to the consumers. It is also a “fencing” point for the transaction \- any messages produced to the same transaction after this marker will be rejected.
+`<txn>-commit` is the commit marker used for marking the transaction as `COMMITTED` and materializing the messages to the consumers. It is also a “fencing” point for the transaction - any messages produced to the same transaction after this marker will be rejected.
Since a transaction can be spreading over multiple messages, we need a mechanism for indexing the messages for the transaction. Hence when the materialization happens, the dispatcher knows how to fetch the messages and dispatch them correctly.
@@ -449,15 +237,15 @@ This can be done via MessageDeduplication cursor. Currently the MessageDeduplica
3. When the marker is successfully written, remove the transaction from MessageDeduplication cursor, since the transaction has already been materialized.
4. The transaction can be added to a transaction cache for fast lookup if needed (optional).
-##### Sweep {#sweep}
+##### Sweep <a id="sweep"></a>
-\`\<txn\>-abort\` is the commit marker used for marking the transaction as \`ABORTED\`. Once a transaction is marked as \`ABORTED\`, the messages of the transaction is safe to be removed. However since managed ledger is append-only, there is no way for deleting individual messages from the partition. So the messages are not easy to remove. The messages have to wait until retention kick in, or an additional “compacting” process is required to compact the segments to delete messages of abo [...]
+`<txn>-abort` is the commit marker used for marking the transaction as `ABORTED`. Once a transaction is marked as `ABORTED`, the messages of the transaction is safe to be removed. However since managed ledger is append-only, there is no way for deleting individual messages from the partition. So the messages are not easy to remove. The messages have to wait until retention kick in, or an additional “compacting” process is required to compact the segments to delete messages of aborted tra [...]
-##### Retention {#retention}
+##### Retention <a id="retention"></a>
In current approach, since transactional messages (both committed and aborted messages) are interleaving with normal messages, there are care should be taken by the broker on acknowledgements. As the cursor can not be moved forward if the transaction that a message belongs to is not finalized (committed or aborted) yet.
-#### Changes {#changes}
+#### Changes <a id="changes"></a>
In summary of this approach, it requires to change following components:
@@ -469,195 +257,196 @@ In summary of this approach, it requires to change following components:
In this approach, we might end up touching almost every pieces at Broker.
-#### Discussions {#discussions}
+#### Discussions <a id="discussions"></a>
There are also a few more performance related discussion points:
-Since the transactional messages appended and transaction commits can happen at different time, the messages of a same transaction is not stored continuously (both logically and physically at bookie side). Hence the entry caching behavior can be very randomized. For example, in Figure 2, when reading messages of txn2, it has to jump back to txn2-m1 and then read \`txn2-m1\` and \`txn2-m2\`; then broker moves to read messages of txn1, it has to jump back to read txn1-m1.
+Since the transactional messages appended and transaction commits can happen at different time, the messages of a same transaction is not stored continuously (both logically and physically at bookie side). Hence the entry caching behavior can be very randomized. For example, in Figure 2, when reading messages of txn2, it has to jump back to txn2-m1 and then read `txn2-m1` and `txn2-m2`; then broker moves to read messages of txn1, it has to jump back to read txn1-m1.
Not only that, in this proposal, we are mixing normal messages with transactional messages, it will significantly change the caching behavior for normal messages, which might results in more network I/O for brokers.
-### Sidecar Approach {#sidecar-approach}
+### Sidecar Approach <a id="sidecar-approach"></a>
In contrast to the marker approach, the other approaches can be described as a sidecar approach. Basically,
-- Use a separate \`managed ledger\` as the “transaction buffer”. Let’s use \`transaction log\` in the renaming discussion of this section.
+- Use a separate `managed ledger` as the “transaction buffer”. Let’s use `transaction log` in the renaming discussion of this section.
- Write the messages involved in transactions to the transaction log.
- Since the messages are not written to the partition, so they are invisible to the consumers until they are committed.
-- When committing a transaction, write a \`COMMIT\` message to the partition, including the list of message ids of its messages and then write the same \`COMMIT\` message to the transaction log. We will discuss why we need two \`COMMIT\` messages in further sections.
-- When aborting a transaction, write a \`ABORT\` message to the transaction log to mark the transaction as aborted. A background process cleans up the messages of aborted transactions.
+- When committing a transaction, write a `COMMIT` message to the partition, including the list of message ids of its messages and then write the same `COMMIT` message to the transaction log. We will discuss why we need two `COMMIT` messages in further sections.
+- When aborting a transaction, write a `ABORT` message to the transaction log to mark the transaction as aborted. A background process cleans up the messages of aborted transactions.
+
+#### Overview <a id="overview-1"></a>
-#### Overview {#overview-1}
+
-![][image3]
-Figure 3\. Sidecar Approach
+Figure 3. Sidecar Approach
-Figure 3 [Sidecar Approach](#bookmark=id.w171cbz337cr) demonstrates how a sidecar approach looks like. The gray boxes represents the messages produced by a normal client (through non-transactional flow); the color boxes represents the messages produced by transactions; different color indicates different transaction. Each message produced by a transaction will be labelled as \`\<txn\>-\<msg\>\` (e.g. \`txn2-m2\`). \`\<txn\>-commit\` or \`\<txn\>-abort\` are the markers appended when comm [...]
+The above figure demonstrates how a sidecar approach looks like. The gray boxes represents the messages produced by a normal client (through non-transactional flow); the color boxes represents the messages produced by transactions; different color indicates different transaction. Each message produced by a transaction will be labelled as `<txn>-<msg>` (e.g. `txn2-m2`). `<txn>-commit` or `<txn>-abort` are the markers appended when committing or aborting a given transaction.
In this approach, all the transactional messages are appended directly to the partition’s transaction log. Each transaction will be using TxnID as the producer id to produce to the partition’s transaction log, so broker can use the de-duplication logic to ensure messages are appended to the transaction log exactly-once.
-1. When transaction coordinator begins committing a transaction, it writes a \`\<txn\>-commit\` marker to the partition to indicate the transaction is marked ast \`COMMITTED\`. This operation seals the transaction. And the commit marker will be served as an index to the transaction messages when broker dispatcher sees it. The message deduplication cursor will have to be improved to involve both transaction log and partition to make sure the commit marker is written exactly once.
-2. When transaction coordinator begins aborting a transaction, it writes a \`\<txn\>-abort\` marker to the transaction log to indicate the transaction is marked as \`ABORTED\`. This operation seals the transaction. After that, the messages of this transaction in transaction log can be cleaned up in the background.
+1. When transaction coordinator begins committing a transaction, it writes a `<txn>-commit` marker to the partition to indicate the transaction is marked ast `COMMITTED`. This operation seals the transaction. And the commit marker will be served as an index to the transaction messages when broker dispatcher sees it. The message deduplication cursor will have to be improved to involve both transaction log and partition to make sure the commit marker is written exactly once.
+2. When transaction coordinator begins aborting a transaction, it writes a `<txn>-abort` marker to the transaction log to indicate the transaction is marked as `ABORTED`. This operation seals the transaction. After that, the messages of this transaction in transaction log can be cleaned up in the background.
-Comparing to the [Marker Approach](#bookmark=id.llk60f1xr4ae), only the \`commit\` marker is written to the partition, so the dispatcher and retention will \*almost\* not be changed. the \`commit\` marker is just a pointer to a batch of messages. The separation of transactional data and normal data will make sure :
+Comparing to the [Marker Approach](#marker-approach), only the `commit` marker is written to the partition, so the dispatcher and retention will *almost* not be changed. the `commit` marker is just a pointer to a batch of messages. The separation of transactional data and normal data will make sure :
1) Transactional use cases will not impact normal use cases
2) It creates an isolation between transactions and normal use cases. Especially around caching perspective. The caching behavior for normal use case will remain unchanged. We can create an enhanced entry cache to optimize it for transaction messages access.
-Figure 3 demonstrates 3 transactions, \`txn1\`, \`txn2\`, and \`txn3\`. \`txn1\` and \`txn2\` are committed, while \`txn3\` is aborted.
+Figure 3 demonstrates 3 transactions, `txn1`, `txn2`, and `txn3`. `txn1` and `txn2` are committed, while `txn3` is aborted.
-#### Challenges {#challenges-1}
+#### Challenges <a id="challenges-1"></a>
There are a few challenges in this approach.
-##### Materialization {#materialization-1}
+##### Materialization <a id="materialization-1"></a>
-Similar as Marker Approach, we use the commit marker \`\<txn\>-commit\` to mark a transaction as \`COMMITTED\` to materialize the messages to the consumers.
+Similar as Marker Approach, we use the commit marker `<txn>-commit` to mark a transaction as `COMMITTED` to materialize the messages to the consumers.
-##### Sweep {#sweep-1}
+##### Sweep <a id="sweep-1"></a>
-\`\<txn\>-abort\` is the commit marker used for marking the transaction as \`ABORTED\`. Once a transaction is marked as \`ABORTED\`, the messages of the transaction is safe to be removed. However since the transaction log is append-only, there is no way for deleting individual messages from the partition. An additional \`compacting\` process is required to be running at background to compact the transaction log to remove the messages of aborted transactions.
+`<txn>-abort` is the commit marker used for marking the transaction as `ABORTED`. Once a transaction is marked as `ABORTED`, the messages of the transaction is safe to be removed. However since the transaction log is append-only, there is no way for deleting individual messages from the partition. An additional `compacting` process is required to be running at background to compact the transaction log to remove the messages of aborted transactions.
-##### Retention {#retention-1}
+##### Retention <a id="retention-1"></a>
Retention becomes much easier comparing to the marker approach. When an acknowledgement happens on a commit marker, it will load the commit marker into the memory and find the message ids of the transaction to acknowledge. Then it will mark those messages as acknowledged in transaction log.
-#### Changes {#changes-1}
+#### Changes <a id="changes-1"></a>
In summary of this approach, it requires to change following components:
-- Introduce a new meta message \- the meta message has a list of pointers to other messages. The dispatcher will resolve the meta message to read the actual messages by following the pointers.
+- Introduce a new meta message - the meta message has a list of pointers to other messages. The dispatcher will resolve the meta message to read the actual messages by following the pointers.
- Change MessageDeduplication to maintain a map between transaction id and its list of message ids, and handle cursors for both transaction log and the partition.
- Introduce a new compacting logic to compact a segment to discard the messages of abort transactions
-#### Discussions {#discussions-1}
+#### Discussions <a id="discussions-1"></a>
-##### Huge Size Transaction {#huge-size-transaction}
+##### Huge Size Transaction <a id="huge-size-transaction"></a>
If we modeled supporting unlimited size message as a transaction of a sequence of message chunks, we can introduce a setting to tell broker to use a separate ledger for storing the messages for a given transaction on a partition. In this approach, we can just let the commit marker to point to the ledger directly. The deletion of the ledger will be similarly as deleting the messages when the commit marker is deleted.
---
-The detailed implementation of the Sidecar Approach is described at Section “[Broker \- Transaction Buffer](#bookmark=id.atfdy9rs9x26)”.
+The detailed implementation of the Sidecar Approach is described at Section “[Broker - Transaction Buffer](#broker---transaction-buffer)”.
-# A Full Proposal {#a-full-proposal}
+# A Full Proposal <a id="a-full-proposal"></a>
In the rest of this design doc we will provide a detailed description of the above transactional flow along with the proposed changes on different modules.
-## Public APIs {#public-apis}
+## Public APIs <a id="public-apis"></a>
-We will introduce a new public API named \`Transaction\`. Transaction is for representing a txn in Pulsar.
+We will introduce a new public API named `Transaction`. Transaction is for representing a txn in Pulsar.
-\`\`\`
-*interface Transaction {*
+```java
+interface Transaction {
- *CompletableFuture\<Void\> commit();*
- *CompletableFuture\<Void\> abort();*
+ CompletableFuture<Void> commit();
+ CompletableFuture<Void> abort();
-*}*
-\`\`\`
+}
+```
---
We will then add a new public API to the PulsarClient interface, and describe how these APIs will be implemented.
-\`\`\`
-/\* initialize a new transaction in pulsar client \*/
-*TransactionBuilder newTransaction();*
-\`\`\`
+```java
+/* initialize a new transaction in pulsar client */
+TransactionBuilder newTransaction();
+```
The *newTransaction* API is used for building a transaction.
-\`\`\`
+```java
-*interface TransactionBuilder {*
+interface TransactionBuilder {
- */\*\**
- *\* Configure the maximum amount of time that the transaction*
- *\* coordinator will for a transaction to be completed by the*
- *\* client before proactively aborting the ongoing transaction.*
- *\**
- *\* The config value will be sent to the transaction coordinator*
- *\* along with the* [CommandNewTxn](#bookmark=id.gtfkqevwo97j)*.*
- *\**
- *\* Default is 60 seconds.*
- *\*/*
- *TransactionBuilder withTransactionTimeout(long timeout, TimeUnit timeoutUnit)*
+ /**
+ * Configure the maximum amount of time that the transaction
+ * coordinator will for a transaction to be completed by the
+ * client before proactively aborting the ongoing transaction.
+ *
+ * The config value will be sent to the transaction coordinator
+ * along with the CommandNewTxn.
+ *
+ * Default is 60 seconds.
+ */
+ TransactionBuilder withTransactionTimeout(long timeout, TimeUnit timeoutUnit)
- */\*\**
- *\* Build the transaction with the configured settings.*
- *\*/*
- *CompletableFuture\<Transaction\> build();*
+ /**
+ * Build the transaction with the configured settings.
+ */
+ CompletableFuture<Transaction> build();
-*}*
-\`\`\`
+}
+```
-The following steps will be taken when is *TransactionBuilder\#build()* called.
+The following steps will be taken when is *TransactionBuilder#build()* called.
1. The pulsar client will discover an available transaction coordinator. It can be any TC alive in the cluster.
-2. Send a [CommandNewTxn](#bookmark=id.gtfkqevwo97j) with configured transaction settings (such Txn TTL time and etc) to the TC. Satisfy the CompletableFuture after retrieving a [TxnID](#bookmark=id.kp4mw4ra0fyl).
+2. Send a [CommandNewTxn](#commandnewtxn) with configured transaction settings (such Txn TTL time and etc) to the TC. Satisfy the CompletableFuture after retrieving a [TxnID](#txnid).
The *Transaction* instance returned from *newTransaction* call will then be used in PulsarClient for producing messages and sending acknowledgements.
---
-We will add an overridden method \`*newMessage*\` to \`*Producer*\` to construct a message to be published as part of a transaction.
+We will add an overridden method `*newMessage*` to `*Producer*` to construct a message to be published as part of a transaction.
-\`\`\`
-/\*\*
- \* Create a message builder to build a message to produce
- \* as part of a transaction
- \*/
-TypedMessageBuilder\<T\>[^4] newMessage(Transaction transaction);
-\`\`\`
+```java
+/**
+ * Create a message builder to build a message to produce
+ * as part of a transaction
+ */
+TypedMessageBuilder<T> newMessage(Transaction transaction);
+```
-With an ongoing transaction, the transaction instance will be responsible for maintaining a set of partitions that it has produced to. When TypedMessageBuilder\#send() is called, the following steps will be added:
+With an ongoing transaction, the transaction instance will be responsible for maintaining a set of partitions that it has produced to. When TypedMessageBuildersend() is called, the following steps will be added:
-1. Check whether the transaction is ongoing. If so, check if the destination partition is in the list of produced partitions. If the partition is not in the list, then send an [CommandAddPartitionToTxn](#bookmark=id.g8xm098glzk9) to TC.
+1. Check whether the transaction is ongoing. If so, check if the destination partition is in the list of produced partitions. If the partition is not in the list, then send an [CommandAddPartitionToTxn](#commandaddpartitiontotxn) to TC.
2. The message will be written into a message batch maintained in the transaction instance. (similar as normal batching logic)
-3. The message batch will be flushed to the wire only after the partition has been added to TC ([CommandAddPartitionToTxn](#bookmark=id.g8xm098glzk9) returns SUCCESS) and when message buffer is full or batching interval has timed out.
-4. The message batch will be written to the wire with a transaction id. See [CommandSend](#bookmark=id.jl31pl47n0cf).
+3. The message batch will be flushed to the wire only after the partition has been added to TC ([CommandAddPartitionToTxn](#commandaddpartitiontotxn) returns SUCCESS) and when message buffer is full or batching interval has timed out.
+4. The message batch will be written to the wire with a transaction id. See [CommandSend](#commandsend).
---
-We will add an overridden method \`*acknowledge*\` to \`*Consumer*\` to acknowledge a given message as part of a transaction.
+We will add an overridden method `*acknowledge*` to `*Consumer*` to acknowledge a given message as part of a transaction.
-\`\`\`
-/\*\*
- \* Acknowledge the consumption of a single message,
- \* identified by its {@link MessageId}, as part of a transaction
- \*/
+```java
+/**
+ * Acknowledge the consumption of a single message,
+ * identified by its {@link MessageId}, as part of a transaction
+ */
void acknowledge(Transaction transaction, MessageId messageId);
-\`\`\`
+```
-When Consumer\#acknowledge() is called, the following steps will be executed:
+When Consumer#acknowledge() is called, the following steps will be executed:
-1. Check whether the transaction is ongoing. If so, check if the destination partition is in the list of acked partitions. If the partition is not in the list, then send an [CommandAddPartitionToTxn](#bookmark=id.g8xm098glzk9) to TC.
-2. The acknowledgement will be send with the transaction id. See [CommandAck](#bookmark=kix.ur9qp3cns1s7).
+1. Check whether the transaction is ongoing. If so, check if the destination partition is in the list of acked partitions. If the partition is not in the list, then send an [CommandAddPartitionToTxn](#commandaddpartitiontotxn) to TC.
+2. The acknowledgement will be send with the transaction id. See [CommandAck](#commandack).
---
-When Transaction\#commit()is called, the following steps will be executed:
+When Transaction#commit()is called, the following steps will be executed:
1. Flush all the pending messages and acknowledges to make sure all modifications made in this transactions are persisted.
2. Send an CommandEndTxn with COMMIT flag to TC, block until the corresponding response is received.
-When Transaction\#abort() is called, the following steps will be executed:
+When Transaction#abort() is called, the following steps will be executed:
1. Immediately fail and drop any buffered messages and acknowledges. Await any outstanding requests which haven’t been acknowledged.
2. Send an CommandEndTxn with ABORT flag to TC, block until the corresponding response is received.
-### Examples {#examples}
+### Examples <a id="examples"></a>
Below is an example on showing how to use Pulsar’s transactional API.
-\`\`\`
-PulsarClient client \= ...;
-Transaction txn \= client.newTransaction().build().get();
+```java
+PulsarClient client = ...;
+Transaction txn = client.newTransaction().build().get();
try {
- Consumer c \= ...;
+ Consumer c = ...;
- Producer p1 \= ...;
- Producer p2 \= ...;
+ Producer p1 = ...;
+ Producer p2 = ...;
- Message msg \= c.receive();
+ Message msg = c.receive();
// send message to p1 in txn
p1.newMessage(txn)
@@ -679,260 +468,260 @@ try {
txn.abort().get();
}
-\`\`\`
+```
-## Data Structures {#data-structures}
+## Data Structures <a id="data-structures"></a>
-#### TxnID {#txnid}
+#### TxnID <a id="txnid"></a>
The TxnID (transaction id) is identifying a unique transaction in pulsar. The transaction id will be 128-bits. The highest 16 bits will be reserved for the id of transaction coordinator, and the remaining bits will be used for monotonically increasing numbers in each TC. so given a TxnID, we can quickly locate the TC.
-## Wire Protocol Changes {#wire-protocol-changes}
+## Wire Protocol Changes <a id="wire-protocol-changes"></a>
-### New Commands {#new-commands}
+### New Commands <a id="new-commands"></a>
-#### TxnAction {#txnaction}
+#### TxnAction <a id="txnaction"></a>
-This is a enum indicating the action for a [*CommandEndTxn*](#bookmark=kix.hoewy3ptkbwj) request.
+This is a enum indicating the action for a [*CommandEndTxn*](#commandendtxn) request.
-\`\`\`
-*enum TxnAction {*
- *COMMIT \= 0;*
- *ABORT \= 1;*
-*}*
-\`\`\`
+```protobuf
+enum TxnAction {
+ COMMIT = 0;
+ ABORT = 1;
+}
+```
-#### CommandNewTxn {#commandnewtxn}
+#### CommandNewTxn <a id="commandnewtxn"></a>
This is a request sent from pulsar client to transaction coordinator when opening a new transaction.
-\`\`\`
-*message CommandNewTxn {*
- *required uint64 request\_id \= 1;*
- *optional uint64 txn\_ttl\_seconds \= 2 \[default \= 0\];*
-*}*
-\`\`\`
+```protobuf
+message CommandNewTxn {
+ required uint64 request_id = 1;
+ optional uint64 txn_ttl_seconds = 2 [default = 0];
+}
+```
-#### CommandNewTxnResponse {#commandnewtxnresponse}
+#### CommandNewTxnResponse <a id="commandnewtxnresponse"></a>
-This is a response for [*CommandNewTxn*](#bookmark=id.gtfkqevwo97j), sent from transaction coordinator to pulsar client.
+This is a response for [*CommandNewTxn*](#commandnewtxn), sent from transaction coordinator to pulsar client.
-\`\`\`
-*message CommandNewTxnResponse {*
- *required uint64 request\_id \= 1;*
- *optional uint64 txnid\_least\_bits \= 2 \[default \= 0\];*
- *optional uint64 txnid\_most\_bits \= 3 \[default \= 0\];*
-*}*
-\`\`\`
+```protobuf
+message CommandNewTxnResponse {
+ required uint64 request_id = 1;
+ optional uint64 txnid_least_bits = 2 [default = 0];
+ optional uint64 txnid_most_bits = 3 [default = 0];
+}
+```
-#### CommandAddPartitionToTxn {#commandaddpartitiontotxn}
+#### CommandAddPartitionToTxn <a id="commandaddpartitiontotxn"></a>
This is a request sent from pulsar client to transaction coordinator to add a partition to the transaction.
-\`\`\`
-*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;*
-*}*
-\`\`\`
+```protobuf
+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;
+}
+```
-#### CommandAddPartitionToTxnResponse {#commandaddpartitiontotxnresponse}
+#### CommandAddPartitionToTxnResponse <a id="commandaddpartitiontotxnresponse"></a>
-This is a response for [*CommandAddPartitionToTxn*](#bookmark=id.g8xm098glzk9), sent from transaction coordinator to pulsar client.
+This is a response for [*CommandAddPartitionToTxn*](#commandaddpartitiontotxn), sent from transaction coordinator to pulsar client.
-\`\`\`
-*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;*
-*}*
-\`\`\`
+```protobuf
+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;
+}
+```
-#### CommandAddSubscritpionToTxn {#commandaddsubscritpiontotxn}
+#### CommandAddSubscritpionToTxn <a id="commandaddsubscritpiontotxn"></a>
This is a request sent from pulsar client to transaction coordinator to add a subscription to the transaction.
-\`\`\`
-*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;*
-*}*
-\`\`\`
-
-#### CommandAddSubscriptionToTxnResponse {#commandaddsubscriptiontotxnresponse}
-
-This is a response for [*CommandAddSubscriptionToTxn*](#bookmark=kix.pwvej5jcgbvs), sent from transaction coordinator to pulsar client.
-
-\`\`\`
-*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;*
-*}*
-\`\`\`
-
-#### CommandEndTxn {#commandendtxn}
+```protobuf
+message Subscription {
+ required string topic = 1;
+ required string subscription = 2;
+}
+```
+
+```protobuf
+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;
+}
+```
+
+#### CommandAddSubscriptionToTxnResponse <a id="commandaddsubscriptiontotxnresponse"></a>
+
+This is a response for [*CommandAddSubscriptionToTxn*](#commandaddsubscritpiontotxn), sent from transaction coordinator to pulsar client.
+
+```protobuf
+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;
+}
+```
+
+#### CommandEndTxn <a id="commandendtxn"></a>
This is a request sent from pulsar client to transaction coordinator when ending (committing or aborting) a new transaction.
-\`\`\`
-*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;*
-*}*
-\`\`\`
+```protobuf
+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;
+}
+```
-#### CommandEndTxnResponse {#commandendtxnresponse}
+#### CommandEndTxnResponse <a id="commandendtxnresponse"></a>
-This is a response for [*CommandEndTxn*](#bookmark=kix.7mfmb7y7zheu), sent from transaction coordinator to pulsar client.
+This is a response for [*CommandEndTxn*](#commandendtxn), sent from transaction coordinator to pulsar client.
-\`\`\`
-*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;*
-*}*
-\`\`\`
+```protobuf
+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;
+}
+```
-#### CommandEndTxnOnPartition {#commandendtxnonpartition}
+#### CommandEndTxnOnPartition <a id="commandendtxnonpartition"></a>
This is a request sent from transaction coordinator to brokers when ending (committing or aborting) a new transaction on a given partition.
-\`\`\`
-*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;*
-*}*
-\`\`\`
+```protobuf
+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;
+}
+```
-#### CommandEndTxnOnPartitionResponse {#commandendtxnonpartitionresponse}
+#### CommandEndTxnOnPartitionResponse <a id="commandendtxnonpartitionresponse"></a>
-This is a response for [*CommandEndTxnOnPartition*](#bookmark=id.ue4e7wrfso8b), sent from brokers back to transaction coordinator.
+This is a response for [*CommandEndTxnOnPartition*](#commandendtxnonpartition), sent from brokers back to transaction coordinator.
-\`\`\`
-*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;*
-*}*
-\`\`\`
+```protobuf
+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;
+}
+```
-#### CommandEndTxnOnSubscription {#commandendtxnonsubscription}
+#### CommandEndTxnOnSubscription <a id="commandendtxnonsubscription"></a>
This is a request sent from transaction coordinator to brokers when ending (committing or aborting) a new transaction on a given subscription.
-\`\`\`
-*message CommandEndTxnOnSubscription {*
- *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 string subscription \= 5;*
- *optional TxnAction txn\_action \= 6;*
-*}*
-\`\`\`
+```protobuf
+message CommandEndTxnOnSubscription {
+ 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 string subscription = 5;
+ optional TxnAction txn_action = 6;
+}
+```
-#### CommandEndTxnOnSubscriptionResponse {#commandendtxnonsubscriptionresponse}
+#### CommandEndTxnOnSubscriptionResponse <a id="commandendtxnonsubscriptionresponse"></a>
-This is a response for [*CommandEndTxnOnSubscription*](#bookmark=kix.we4rd1tv1zpn), sent from brokers back to transaction coordinator.
+This is a response for [*CommandEndTxnOnSubscription*](#commandendtxnonsubscription), sent from brokers back to transaction coordinator.
-\`\`\`
-*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;*
-*}*
-\`\`\`
+```protobuf
+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;
+}
+```
-### Modified Commands {#modified-commands}
+### Modified Commands <a id="modified-commands"></a>
-#### CommandSend {#commandsend}
+#### CommandSend <a id="commandsend"></a>
This is a request sent from pulsar clients to brokers to append messages.
-\`\`\`
-*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\];***
-*}*
-\`\`\`
+```protobuf
+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];
+}
+```
-#### CommandAck {#commandack}
+#### CommandAck <a id="commandack"></a>
This is a request sent from pulsar clients to brokers to acknowledge messages.
-\`\`\`
-*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\];***
-*}*
-
-\`\`\`
-
-\`\`\`
-*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;*
-*}*
-\`\`\`
-
-## Transaction Coordinator {#transaction-coordinator-1}
+```protobuf
+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];
+}
+
+```
+
+```protobuf
+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;
+}
+```
+
+## Transaction Coordinator <a id="transaction-coordinator-1"></a>
The transaction coordinator is a module used for handling requests from the transactional pulsar clients to keep track of their transaction status. Each transaction coordinator is identified by an id (TCID), and maintains its own transaction metadata store. TCID is used for generating transaction id, as well as for recovery from TC failures.
-The operations of the transaction metadata store can be abstracted into a \`*TransactionMetadataStore*\` and \`*TransactionMetadataStoreProvider\`* interface.
+The operations of the transaction metadata store can be abstracted into a `*TransactionMetadataStore*` and `*TransactionMetadataStoreProvider`* interface.
-\`\`\`
+```java
enum TxnStatus {
OPEN,
COMMITTING,
@@ -943,56 +732,56 @@ enum TxnStatus {
interface TransactionMetadataStoreProvider {
- /\*\*
- \* Open the transaction metadata store for transaction coordinator
- \* identified by \<tt\>transactionCoordinatorId\</tt\>.
- \*/
- CompletableFuture\<TransactionMetadataStore\> openStore(
+ /**
+ * Open the transaction metadata store for transaction coordinator
+ * identified by transactionCoordinatorId.
+ */
+ CompletableFuture<TransactionMetadataStore> openStore(
long transactionCoordinatorId);
}
interface TransactionMetadataStore {
- /\*\*
- \* Create a new transaction in the transaction metadata store.
- \*/
- CompletableFuture\<TxnID\> newTransaction();
-
- /\*\*
- \* Add the produced partitions to transaction identified by \<tt\>txnid\</tt\>.
- \*/
- CompletableFture\<Void\> addProducedPartitionToTxn(
- TxnID txnid, List\<String\> partitions);
-
- /\*\*
- \* Add the acked partitions to transaction identified by \<tt\>txnid\</tt\>.
- \*/
- CompletableFture\<Void\> addAckedPartitionToTxn(
- TxnID txnid, List\<String\> partitions);
-
- /\*\*
- \* Update the transaction from \<tt\>expectedStatus\</tt\> to \<tt\>newStatus\</tt\>.
- \*
- \* \<p\>If the current transaction status is not \<tt\>expectedStatus\</tt\>, the
- \* update will be failed.
- \*/
- CompletableFuture\<Void\> updateTxnStatus(
+ /**
+ * Create a new transaction in the transaction metadata store.
+ */
+ CompletableFuture<TxnID> newTransaction();
+
+ /**
+ * Add the produced partitions to transaction identified by txnid.
+ */
+ CompletableFture<Void> addProducedPartitionToTxn(
+ TxnID txnid, List<String> partitions);
+
+ /**
+ * Add the acked partitions to transaction identified by txnid.
+ */
+ CompletableFture<Void> addAckedPartitionToTxn(
+ TxnID txnid, List<String> partitions);
+
+ /**
+ * Update the transaction from expectedStatus to newStatus.
+ *
+ * If the current transaction status is not expectedStatus, the
+ * update will be failed.
+ */
+ CompletableFuture<Void> updateTxnStatus(
TxnStatus newStatus, TxnStatus expectedStatus);
}
-\`\`\`
+```
-### Concepts {#concepts-1}
+### Concepts <a id="concepts-1"></a>
-#### TCID {#tcid}
+#### TCID <a id="tcid"></a>
Transaction Coordinator ID (TCID) is used for identifying an instance of transaction coordinator. This TCID is used for locating the transaction metadata store served by this transaction coordinator.
-We will discuss on how to run transaction coordinators and how to assign TCID at Section [Deployment](#bookmark=id.kqdalafxhiah).
+We will discuss on how to run transaction coordinators and how to assign TCID at Section [Deployment](#deployment).
-#### TxnStatus {#txnstatus}
+#### TxnStatus <a id="txnstatus"></a>
-\`TxnStatus\` is used for indicating the status of a transaction.
+`TxnStatus` is used for indicating the status of a transaction.
- OPEN: The transaction is open. The operations of the transaction is ongoing.
- COMMITTING: TC is committing the transaction.
@@ -1000,17 +789,17 @@ We will discuss on how to run transaction coordinators and how to assign TCID at
- ABORTING: TC is aborting the transaction.
- ABORTED: The transaction has been aborted. The metadata of the transaction can be removed from the transaction metadata store at this moment.
-#### TransactionMetadataStoreProvider {#transactionmetadatastoreprovider}
+#### TransactionMetadataStoreProvider <a id="transactionmetadatastoreprovider"></a>
The provider is used for locating and opening the TransactionMetadataStore for a given transaction coordinator (identified by TCID).
-#### TransactionMetadataStore {#transactionmetadatastore}
+#### TransactionMetadataStore <a id="transactionmetadatastore"></a>
The transaction metadata store is used by a transaction coordinator for storing and managing all the transactions’ metadata.
The metadata store should provide following implementations for the interfaces described above.
-### Implementation {#implementation}
+### Implementation <a id="implementation"></a>
The transaction metadata store can be implemented by using a distributed key/value storage (like zookeeper), or using an in-memory hashmap backed by a managed ledger.
@@ -1023,41 +812,41 @@ In this implementation, the transaction coordinator maintains the following info
All the updates to the transaction metadata map will first be persisted to the transaction log (backed by a managed ledger). So they can be used for recovery.
-#### Transaction Log {#transaction-log}
+#### Transaction Log <a id="transaction-log"></a>
The format of the transaction entry appended to the transaction log is described as below.
-\`\`\`
-*enum TransactionMetadataOp {*
- *NEW \= 0;*
- *ADD\_PARTITION \= 1;*
- *ADD\_SUBSCRIPTION \= 2;*
- *UPDATE \= 3;*
-*}*
+```protobuf
+enum TransactionMetadataOp {
+ NEW = 0;
+ ADD_PARTITION = 1;
+ ADD_SUBSCRIPTION = 2;
+ UPDATE = 3;
+}
-*message TransactionMetadataEntry {*
- *TransactionMetadataOp metadata\_op \= 1;*
- *optional uint64 txnid\_least\_bits \= 2 \[default \= 0\];*
- *optional uint64 txnid\_most\_bits \= 3 \[default \= 0\];*
- *optional TxnStatus expected\_status \= 4;*
- *optional TxnStatus new\_status \= 5;*
- *repeated string partitions \= 6;*
- *repeated Subscription subscriptions \= 7;*
- *optional uint64 txn\_timeout\_ms \= 8;*
- *optional uint64 txn\_start\_time \= 9;*
- *optional uint64 txn\_last\_modification\_time \= 10;*
-*}*
-\`\`\`
+message TransactionMetadataEntry {
+ TransactionMetadataOp metadata_op = 1;
+ optional uint64 txnid_least_bits = 2 [default = 0];
+ optional uint64 txnid_most_bits = 3 [default = 0];
+ optional TxnStatus expected_status = 4;
+ optional TxnStatus new_status = 5;
+ repeated string partitions = 6;
+ repeated Subscription subscriptions = 7;
+ optional uint64 txn_timeout_ms = 8;
+ optional uint64 txn_start_time = 9;
+ optional uint64 txn_last_modification_time = 10;
+}
+```
-The *TransactionMetadataOp* field indicating the operations happened in the transaction metadata store; while the *expected\_status* indicating the expected txn status for this operation and *new\_status* indicating the txn status should be updated.
+The *TransactionMetadataOp* field indicating the operations happened in the transaction metadata store; while the *expected_status* indicating the expected txn status for this operation and *new_status* indicating the txn status should be updated.
-Writing COMMITTING or ABORTING transaction metadata entry can be served as the synchronization point \- Once the transaction metadata entry is appended to the transaction log, the transaction is GUARANTEED to be committed or aborted. Even when the TC fails, the transaction will be rolled forward or back upon recovery.
+Writing COMMITTING or ABORTING transaction metadata entry can be served as the synchronization point - Once the transaction metadata entry is appended to the transaction log, the transaction is GUARANTEED to be committed or aborted. Even when the TC fails, the transaction will be rolled forward or back upon recovery.
-The timestamp in the transaction metadata entry will be used for determining when the transaction has timed out. Once the difference of the current time and the *txn\_start\_time* has exceeded the *txn\_timeout\_ms* value, the transaction will be aborted.
+The timestamp in the transaction metadata entry will be used for determining when the transaction has timed out. Once the difference of the current time and the *txn_start_time* has exceeded the *txn_timeout_ms* value, the transaction will be aborted.
-#### TC Startup {#tc-startup}
+#### TC Startup <a id="tc-startup"></a>
-The TC will be assigned with a TCID during startup. The TCID assignment will be determined by the deployment method used for deploying transaction coordinators. The TC will basically call \`TransactionMetadataStoreProvider\#openStore(TCID)\` to open its transaction metadata store. The execution of \`openStore\` is described as followings:
+The TC will be assigned with a TCID during startup. The TCID assignment will be determined by the deployment method used for deploying transaction coordinators. The TC will basically call `TransactionMetadataStoreProvider#openStore(TCID)` to open its transaction metadata store. The execution of `openStore` is described as followings:
1. Open its currently assigned managed ledger (identified by TCID).
2. Bootstrap the transaction metadata cache: TC will scan the managed ledger from the beginning[^5] to replay the entries and materialize the entries to the metadata cache.
@@ -1066,23 +855,23 @@ During replaying, there are a few actions should be taken care of:
When **committing** a transaction, the following steps should be done by TC:
-1. Send [CommandEndTxnOnPartition](#bookmark=id.ue4e7wrfso8b) with *COMMIT* action to all the brokers of the transaction’s added partitions.
-2. Send [CommandEndTxnOnSubscription](#bookmark=kix.we4rd1tv1zpn) with *COMMIT* action to all the brokers of the transaction’s added subscriptions.
-3. When all the responses have been received, append a [*TransactionMetadataEntry*](#bookmark=id.iasxbbpsygfk) to its transaction log to change the transaction status to *COMMITTED*.
+1. Send [CommandEndTxnOnPartition](#commandendtxnonpartition) with *COMMIT* action to all the brokers of the transaction’s added partitions.
+2. Send [CommandEndTxnOnSubscription](#commandendtxnonsubscription) with *COMMIT* action to all the brokers of the transaction’s added subscriptions.
+3. When all the responses have been received, append a [*TransactionMetadataEntry*](#transaction-log) to its transaction log to change the transaction status to *COMMITTED*.
When **aborting** a transaction, the following steps will be executed by TC:
-1. Send [CommandEndTxnOnPartition](#bookmark=id.ue4e7wrfso8b) with *ABORT* action to all the brokers of the transaction’s added partitions.
-2. Send [CommandEndTxnOnSubscription](#bookmark=kix.we4rd1tv1zpn) with *ABORT* action to all the brokers of the transaction’s added subscriptions.
-3. When all the responses have been received, append a [*TransactionMetadataEntry*](#bookmark=id.iasxbbpsygfk) to its transaction log to change the transaction status to *ABORTED*.
+1. Send [CommandEndTxnOnPartition](#commandendtxnonpartition) with *ABORT* action to all the brokers of the transaction’s added partitions.
+2. Send [CommandEndTxnOnSubscription](#commandendtxnonsubscription) with *ABORT* action to all the brokers of the transaction’s added subscriptions.
+3. When all the responses have been received, append a [*TransactionMetadataEntry*](#transaction-log) to its transaction log to change the transaction status to *ABORTED*.
---
-One important thing to keep in mind is \- when ending (committing or aborting) a transaction, if one of the partitions involved in the transaction become unavailable, then the transaction will be unable to completed. This will ONLY delay the materialization of messages from transaction buffer to the actual data partition, but it doesn’t block consumers on consuming already materialized (READ\_COMMITTED) data.
+One important thing to keep in mind is - when ending (committing or aborting) a transaction, if one of the partitions involved in the transaction become unavailable, then the transaction will be unable to completed. This will ONLY delay the materialization of messages from transaction buffer to the actual data partition, but it doesn’t block consumers on consuming already materialized (READ_COMMITTED) data.
-#### TC Request Processing {#tc-request-processing}
+#### TC Request Processing <a id="tc-request-processing"></a>
-##### CommandNewTxn {#commandnewtxn-1}
+##### CommandNewTxn <a id="commandnewtxn-1"></a>
When TC receiving the CommandNewTxn request, the following steps will be executed:
@@ -1091,39 +880,39 @@ When TC receiving the CommandNewTxn request, the following steps will be execute
- An TransactionMetadataEntry with TxnID and OPEN TxnStatus is logged into the transaction log.
- Respond with the TxnID to the client
-##### CommandAddPartitionToTxn {#commandaddpartitiontotxn-1}
+##### CommandAddPartitionToTxn <a id="commandaddpartitiontotxn-1"></a>
When TC receiving the CommandAddPartitionToTxn request, the following steps will be executed:
- If the TxnID does not exist in transaction metadata store, it responds TxnNotFound error; otherwise proceed to next step.
-- If there is already a transaction metadata entry exists in the store, check if the status is OPEN , if yes append a transaction metadata entry with the updated partition list and the updated last\_modification\_time timestamp, respond success when the entry is successfully appended to the log; otherwise reply with TxnMetaStoreError error code.
+- If there is already a transaction metadata entry exists in the store, check if the status is OPEN , if yes append a transaction metadata entry with the updated partition list and the updated last_modification_time timestamp, respond success when the entry is successfully appended to the log; otherwise reply with TxnMetaStoreError error code.
-##### CommandAddSubscriptionToTxn {#commandaddsubscriptiontotxn}
+##### CommandAddSubscriptionToTxn <a id="commandaddsubscriptiontotxn"></a>
When TC receiving the CommandAddSubscriptionToTxn request, the following steps will be executed:
- If the TxnID does not exist in transaction metadata store, it responds TxnNotFound error; otherwise proceed to next step.
-- If there is already a transaction metadata entry exists in the store, check if the status is OPEN , if yes append a transaction metadata entry with the updated subscription list and the updated last\_modification\_time timestamp, respond success when the entry is successfully appended to the log; otherwise reply with TxnMetaStoreError error code.
+- If there is already a transaction metadata entry exists in the store, check if the status is OPEN , if yes append a transaction metadata entry with the updated subscription list and the updated last_modification_time timestamp, respond success when the entry is successfully appended to the log; otherwise reply with TxnMetaStoreError error code.
-##### CommandEndTxn {#commandendtxn-1}
+##### CommandEndTxn <a id="commandendtxn-1"></a>
When TC receiving the CommandEndTxn request, the following steps will be executed:
1. If the TxnID does not exist in transaction metadata store, it responds TxnNotFound error; otherwise proceed to next step.
2. If there is already an entry found in the transaction metadata store, check its status
- 1. If the status is OPEN, go on to step 3\.
+ 1. If the status is OPEN, go on to step 3.
2. If the status is COMMITTING or COMMITTED, and the action from the request is COMMIT, then return success.
3. If the status is ABORTING or ABORTED, and the action from the request is ABORT, then return success.
4. Otherwise, reply with TxnMetaStoreError error code.
3. Construct a transaction metadata entry with updating following fields:
- 1. Update last\_modification\_time
+ 1. Update last_modification_time
2. Change TxnStatus according to the action from the request
1. If action is COMMIT action, change TxnStatus to COMMITTING.
2. If action is ABORT action, change TxnStatus to ABORTING.
4. Append the transaction metadata entry to the transaction log, and update the in-memory map upon successfully append the entry, and respond success to the client.
5. TC commits or aborts the transaction based on the action from the request.
-#### TC Transaction Expiration {#tc-transaction-expiration}
+#### TC Transaction Expiration <a id="tc-transaction-expiration"></a>
When a pulsar client crashes and never comes back, TC should have a mechanism to proactively expire the transactions initiated by this client. To achieve this, TC will periodically scan the transaction metadata store:
@@ -1133,7 +922,7 @@ When a pulsar client crashes and never comes back, TC should have a mechanism to
2. If a transaction is in COMMITTING state, then complete the commiting process for this transaction. (This can happen when a TC crashes at the middle of completing the committing process)
3. If a transaction is in ABORTING state, then complete the aborting process for this transaction. (This can happen when a TC crashes at the middle of completing the aborting process)
-### Deployment {#deployment}
+### Deployment <a id="deployment"></a>
Transaction Coordinator (TC) is designed to be a logic component, which can be deployed into multiple instances. The main question of TC is how to assign a unique TCID when deploying TCs.
@@ -1143,67 +932,67 @@ If you deploy TC as a Pulsar Function, you can obtain the instance id as the TCI
If you are running TC as part of a broker, this can be done in a different way, described in the below section.
-#### TC in Broker {#tc-in-broker}
-
-We can model all the transaction logs for TCs as a \`TC-Transaction-Log\` topic. The number of TCs determines the number of partitions of this transaction log topic. Each TC will be obtaining one partition to startup. Since partitions area already managed and distributed to brokers by load manager. So when a partition of the TC transaction log topic is assigned to a broker, the broker knows the partition id and can use the partition id as TCID to start up.
+#### TC in Broker <a id="tc-in-broker"></a>
-## Broker \- Transaction Buffer {#broker---transaction-buffer}
+We can model all the transaction logs for TCs as a `TC-Transaction-Log` topic. The number of TCs determines the number of partitions of this transaction log topic. Each TC will be obtaining one partition to startup. Since partitions area already managed and distributed to brokers by load manager. So when a partition of the TC transaction log topic is assigned to a broker, the broker knows the partition id and can use the partition id as TCID to start up.
-This proposal implements a [Sidecar Approach](#bookmark=id.x2y17hd6i2md) described above.
+## Broker - Transaction Buffer <a id="broker---transaction-buffer"></a>
-### Message {#message}
+This proposal implements a [Sidecar Approach](#sidecar-approach) described above.
-\`\`\`
-*enum MessageType {*
+### Message <a id="message"></a>
- *DATA \= 0;*
- *TXN\_COMMIT \= 1;*
- *TXN\_ABORT \= 2;*
+```protobuf
+enum MessageType {
-*}*
-\`\`\`
+ DATA = 0;
+ TXN_COMMIT = 1;
+ TXN_ABORT = 2;
-\`\`\`
-*message MessageMetadata {*
- *…*
- *optional MessageType msg\_type \= 19 \[default \= DATA\];*
- *optional long indirect\_ledger\_id \= 20;*
- *optional uint64 txnid\_least\_bits \= 21 \[default \= 0\];*
- *optional uint64 txnid\_most\_bits \= 22 \[default \= 0\];*
-*}*
-\`\`\`
+}
+```
+
+```protobuf
+message MessageMetadata {
+ …
+ optional MessageType msg_type = 19 [default = DATA];
+ optional long indirect_ledger_id = 20;
+ optional uint64 txnid_least_bits = 21 [default = 0];
+ optional uint64 txnid_most_bits = 22 [default = 0];
+}
+```
We introduce a new field in message metadata indicating the message type.
- DATA: indicates this is a normal message, which payload is storing the actual payload of the message.
-- TXN\_COMMIT: this is a commit marker message for indicating that a transaction is committed.
-- TXN\_ABORT: this is an abort marker message for indicating that a transaction has been aborted.
+- TXN_COMMIT: this is a commit marker message for indicating that a transaction is committed.
+- TXN_ABORT: this is an abort marker message for indicating that a transaction has been aborted.
-#### Message ID {#message-id}
+#### Message ID <a id="message-id"></a>
The message id of the message in a transaction is comprised by the message id of the commit marker in topic partition and the index of the message in the transaction. For example, if the transaction is committed as ledger id 5, entry id 7. The message id of 2nd message in the transaction is 5,7,2.
-### Transaction Log {#transaction-log-1}
+### Transaction Log <a id="transaction-log-1"></a>
-For each topic partition where a transaction is happening, broker will automatically created a sidecar partition and use it as the transaction log. The name of the transaction log partition is the name of the topic partition with a suffix \`/\_txnlog\`. For example, if the topic partition is \`my\_tenant/my\_ns/my\_topic\`, then the transaction log partition for it is \`my\_tenant/my\_ns/my\_topic/\_txnlog\`.
+For each topic partition where a transaction is happening, broker will automatically created a sidecar partition and use it as the transaction log. The name of the transaction log partition is the name of the topic partition with a suffix `/_txnlog`. For example, if the topic partition is `my_tenant/my_ns/my_topic`, then the transaction log partition for it is `my_tenant/my_ns/my_topic/_txnlog`.
The transaction log partition will only be created when there are transactions. So it doesn’t introduced additional overhead if transactions are not used.
-### Implementation {#implementation-1}
+### Implementation <a id="implementation-1"></a>
-#### Transaction Manager[^6] {#transaction-manager}
+#### Transaction Manager[^6] <a id="transaction-manager"></a>
-The implementation of a transaction manager is based on \`Message Deduplication\` (let’s call it \`dedup\` cursor in the remaining part of this proposal). It comprised a \`dedup\` cursor, a \`retention\` cursor, a \`transaction\` cursor and a transaction log.
+The implementation of a transaction manager is based on `Message Deduplication` (let’s call it `dedup` cursor in the remaining part of this proposal). It comprised a `dedup` cursor, a `retention` cursor, a `transaction` cursor and a transaction log.
-##### Dedup Cursor {#dedup-cursor}
+##### Dedup Cursor <a id="dedup-cursor"></a>
-The Dedup cursor is actually the \`Message Deduplication\` on the transaction log. It ensures for a given transaction, all the messages are appended exactly-once to the transaction log.
+The Dedup cursor is actually the `Message Deduplication` on the transaction log. It ensures for a given transaction, all the messages are appended exactly-once to the transaction log.
-##### Transaction Cursor {#transaction-cursor}
+##### Transaction Cursor <a id="transaction-cursor"></a>
-The transaction cursor maintains a map between TxnID and TransactionIndex. The cursor is used for rebuilding the transaction map during recovery, like how \`dedup\` cursor rebuilds its sequence id maps.
+The transaction cursor maintains a map between TxnID and TransactionIndex. The cursor is used for rebuilding the transaction map during recovery, like how `dedup` cursor rebuilds its sequence id maps.
-###### *Transaction Index* {#transaction-index}
+###### *Transaction Index* <a id="transaction-index"></a>
A transaction index includes following fields:
@@ -1211,17 +1000,17 @@ A transaction index includes following fields:
- List\<MessageId\>: A list of message ids pointing to the messages in the transaction log
- CommitMarker: The message id of the commit marker in the topic partition
-The \`CommitMarker\` can be used for building a reverse index from \`CommitMarker\` to \`Transaction Index\`. If we sort the reverse index by \`CommitMarker\` we can use this reverse index quickly locate the transactions can be deleted when the segments are deleted from topic partition, and delete those transactions from the transaction log.
+The `CommitMarker` can be used for building a reverse index from `CommitMarker` to `Transaction Index`. If we sort the reverse index by `CommitMarker` we can use this reverse index quickly locate the transactions can be deleted when the segments are deleted from topic partition, and delete those transactions from the transaction log.
-##### Retention Cursor {#retention-cursor}
+##### Retention Cursor <a id="retention-cursor"></a>
The retention cursor is used for managing the retention of messages in the transaction log. It is created along with the creation of the transaction log.
-When a \`transaction\` is aborted, it learns the list of messages of a transaction from the \`Transaction Cursor\` and mark the list of messages as deleted.
+When a `transaction` is aborted, it learns the list of messages of a transaction from the `Transaction Cursor` and mark the list of messages as deleted.
-When a segment is deleted from topic partition due to retention, broker find the transactions whose commit markers has been deleted by searching the commit marker reverse index in \`Transaction Cursor\`, and mark the list of messages of those deleted transactions as deleted in transaction log.
+When a segment is deleted from topic partition due to retention, broker find the transactions whose commit markers has been deleted by searching the commit marker reverse index in `Transaction Cursor`, and mark the list of messages of those deleted transactions as deleted in transaction log.
-#### Compacting Transaction Log {#compacting-transaction-log}
+#### Compacting Transaction Log <a id="compacting-transaction-log"></a>
If users set retention to infinite, it is making more sense to compact transaction log to delete the messages of aborted messages and rewrite the messages in transaction commit order.
@@ -1229,19 +1018,18 @@ The compaction is done by using another transaction log and rewriting the transa
The compacting flow is described as below:
-- Create a \`Compacting Cursor\` on transaction log
-- Loop over the commit marker reverse index in \`Transaction Cursor\`
+- Create a `Compacting Cursor` on transaction log
+- Loop over the commit marker reverse index in `Transaction Cursor`
- For each commit transaction
- Read its transactional messages one by one
- Write the transactional messages in order to a new transaction log
- At the completion of the rewrite, update the in-memory transaction index to point to the new locations
- - Update \`Compacting Cursor\` to mark the transactional messages as DELETED. So they can be deleted.
+ - Update `Compacting Cursor` to mark the transactional messages as DELETED. So they can be deleted.
- The broker keeps doing the loop whenever there is a new transaction committed.
-
The same compacting logic can be applied when a segment is offloaded to tiered storage. We will append the transaction messages when offloading the segment.
-#### Partition Loading {#partition-loading}
+#### Partition Loading <a id="partition-loading"></a>
When a partition is loaded, it will check if there is a transaction log. If there is no transaction log, it means that the partition doesn’t have any transaction, it will recover as normal.
@@ -1253,81 +1041,81 @@ If there is a transaction log, it will recover transaction manager as part of th
- Then it will start replay the transaction log to rebuild the transaction index
- After finishing the replay, it completes the recovery.
-#### Broker Request Processing {#broker-request-processing}
+#### Broker Request Processing <a id="broker-request-processing"></a>
-##### CommandSend {#commandsend-1}
+##### CommandSend <a id="commandsend-1"></a>
When TB receiving the CommandSend request with a transaction id, the following steps will be executed:
- It will use the txn id as the producer id to append the message into transaction log, dedup cursor will make sure the message is never duplicated.
-- After successfully appended the message to the transaction log, it will update the transaction index in \`Transaction Cursor\`.
+- After successfully appended the message to the transaction log, it will update the transaction index in `Transaction Cursor`.
-##### CommandEndTxnOnPartition {#commandendtxnonpartition-1}
+##### CommandEndTxnOnPartition <a id="commandendtxnonpartition-1"></a>
When TB receiving the CommandEndTxnOnPartition request with COMMIT action, the following steps will be executed:
-- It will first write a message with message type TXN\_COMMIT with transaction id to the partition.
+- It will first write a message with message type TXN_COMMIT with transaction id to the partition.
- After it successfully appended to the topic, it will get the message id of this commit marker message.
-- It will then write another message with message type TXN\_COMMIT with transaction id to the transaction log. The payload of this message will include the message id of the commit marker in topic partition. It will be used for recovering \`Transaction Cursor\` upon failures.
+- It will then write another message with message type TXN_COMMIT with transaction id to the transaction log. The payload of this message will include the message id of the commit marker in topic partition. It will be used for recovering `Transaction Cursor` upon failures.
-\> We added commit marker to the partition first so we can retrieve the message id of the commit marker and store it in \`Transaction Cursor\`
+\> We added commit marker to the partition first so we can retrieve the message id of the commit marker and store it in `Transaction Cursor`
When TB receiving the CommandEndTxnOnPartition request with ABORT action, the following steps will be executed:
-- It will write a message with message type TXN\_ABORT with transaction id to the transaction log.
-- Then it will mark the transaction as aborted and mark the messages of this transaction as DELETED in \`Retention Cursor\`.
+- It will write a message with message type TXN_ABORT with transaction id to the transaction log.
+- Then it will mark the transaction as aborted and mark the messages of this transaction as DELETED in `Retention Cursor`.
-##### Broker Dispatching {#broker-dispatching}
+##### Broker Dispatching <a id="broker-dispatching"></a>
-When Broker dispatcher dispatches a message \`TXN\_COMMIT\`, it will retrieve the list of messages ids of the transaction from \`Transaction Cursor\`. Then it will read individual messages from the transaction log and dispatch them.
+When Broker dispatcher dispatches a message `TXN_COMMIT`, it will retrieve the list of messages ids of the transaction from `Transaction Cursor`. Then it will read individual messages from the transaction log and dispatch them.
-## Broker \- Acknowledgements {#broker---acknowledgements}
+## Broker - Acknowledgements <a id="broker---acknowledgements"></a>
-As mentioned in Section [Motivation](#bookmark=id.g2xdepks9p63), Pulsar Functions need to both consume from input topics and produce results to output topics at the same time. When Pulsar Functions acknowledges input messages, the acknowledgements should be done as part of the transaction. In order to support this scenario, we have to improve \`Subscription\` component at Pulsar broker side to make it transaction-aware. More specifically, we need to:
+As mentioned in Section [Motivation](#motivation), Pulsar Functions need to both consume from input topics and produce results to output topics at the same time. When Pulsar Functions acknowledges input messages, the acknowledgements should be done as part of the transaction. In order to support this scenario, we have to improve `Subscription` component at Pulsar broker side to make it transaction-aware. More specifically, we need to:
1. Enhance acknowledgement protocol
- 1. Improve \`CommandAck\` to carry transaction id
- 2. Introduce a new wire protocol message for client to receive acknowledgement response \`CommandAckResponse\`.
-2. Introduce a new acknowledgement state \`PENDING\_ACK\` in Subscription for achieving transactional acknowledgement.
+ 1. Improve `CommandAck` to carry transaction id
+ 2. Introduce a new wire protocol message for client to receive acknowledgement response `CommandAckResponse`.
+2. Introduce a new acknowledgement state `PENDING_ACK` in Subscription for achieving transactional acknowledgement.
-### New Acknowledgement State {#new-acknowledgement-state}
+### New Acknowledgement State <a id="new-acknowledgement-state"></a>
-![][image4]
+
Figure. Acknowledgement State Machine
The diagram above shows the state machine of acknowledgement flow at broker side. The black boxes are the states currently Pulsar has, while the blue boxes are the new states introduced by supporting transactional acknowledgement. The black arrow line indicates the state change, while the blue arrow lines indicates state changes introduced after introducing the new state.
-Before introducing the new \`PENDING\_ACK\` state, Pulsar technically has two states for a message in the dispatcher path. One is \`PENDING\`, while the other one is \`DELETED\`. When a message is delivered to a consumer, the message will be put into \`PENDING\`; when broker receives an acknowledgement from a consumer, the message will be marked as \`DELETED\`. If a redelivery request received from consumers, broker will re-deliver the messages in \`PENDING\`.
+Before introducing the new `PENDING_ACK` state, Pulsar technically has two states for a message in the dispatcher path. One is `PENDING`, while the other one is `DELETED`. When a message is delivered to a consumer, the message will be put into `PENDING`; when broker receives an acknowledgement from a consumer, the message will be marked as `DELETED`. If a redelivery request received from consumers, broker will re-deliver the messages in `PENDING`.
-#### PENDING\_ACK State {#pending_ack-state}
+#### PENDING_ACK State <a id="pending_ack-state"></a>
-In order to support transactional acknowledgement, we introduce a new state for messages. This state is called \`PENDING\_ACK\`. When the message is put in this state, it means:
+In order to support transactional acknowledgement, we introduce a new state for messages. This state is called `PENDING_ACK`. When the message is put in this state, it means:
- The message CANNOT be re-delivered to any other consumers even it is requested to re-deliver by consumers.
-- The message can ONLY come out of the PENDING\_ACK state when an explicit EndTxn request is sent to the subscription, via \`[CommandEndTxnOnSubscription](#bookmark=kix.we4rd1tv1zpn)\`.
+- The message can ONLY come out of the PENDING_ACK state when an explicit EndTxn request is sent to the subscription, via [CommandEndTxnOnSubscription](#commandendtxnonsubscription).
-After introduced this \`PENDING\_ACK\` state, the acknowledgement state machine will be changed to following:
+After introduced this `PENDING_ACK` state, the acknowledgement state machine will be changed to following:
-1. PENDING \-\> DELETED: on receiving a normal ACK request. However a new logic will be introduced in this state transition. Before it can transition to \`DELETED\` state, it has to check if the message is in \`PENDING\_ACK\`. If the message is in \`PENDING\_ACK\`, it CANNOT be marked as \`DELETED\`, we can simply drop the acknowledgement on this message[^7]. If the message is not in \`PENDING\_ACK\`, transition the state to \`DELETED\` state as a normal ACK.
-2. PENDING \-\> PENDING\_ACK: when receiving [an ACK request with transaction id](#bookmark=kix.ur9qp3cns1s7), the broker will first check if this message can be transitioned to \`PENDING\_ACK\`. A message can only be transitioned when it is in \`PENDING\` state. If the message is in any other state, the ACK request will be failed with \`InvalidTxnState\` error code. This will result in client aborting the transaction since conflicts are detected[^8].
- When a message is transitioned to \`PENDING\_ACK\` state, broker will log a state change to cursor ledger. This ensures the \`PENDING\_ACK\` state is durably stored and can be recovered upon broker failures.
-3. PENDING\_ACK \-\> PENDING: on receiving \`[CommandEndTxnOnSubscription](#bookmark=kix.we4rd1tv1zpn)\` request with \`ABORT\` action, the broker logs a state change into cursor ledger of message state transitioning from \`PENDING\_ACK\` to \`PENDING\`. This ensures the \`PENDING\_ACK\` state can be cleared and the message can be re-dispatched in next consume request.
-4. PENDING\_ACK \-\> DELETED: on receiving \`[CommandEndTxnOnSubscription](#bookmark=kix.we4rd1tv1zpn)\` request with \`COMMIT\` action, the broker logs a state change into cursor ledger of message state transitioning from \`PENDING\_ACK\` to \`DELETED\`. This ensures the message can be deleted and will not be delivered again.
+1. PENDING -\> DELETED: on receiving a normal ACK request. However a new logic will be introduced in this state transition. Before it can transition to `DELETED` state, it has to check if the message is in `PENDING_ACK`. If the message is in `PENDING_ACK`, it CANNOT be marked as `DELETED`, we can simply drop the acknowledgement on this message[^7]. If the message is not in `PENDING_ACK`, transition the state to `DELETED` state as a normal ACK.
+2. PENDING -\> PENDING_ACK: when receiving [an ACK request with transaction id](#commandack), the broker will first check if this message can be transitioned to `PENDING_ACK`. A message can only be transitioned when it is in `PENDING` state. If the message is in any other state, the ACK request will be failed with `InvalidTxnState` error code. This will result in client aborting the transaction since conflicts are detected[^8].
+ When a message is transitioned to `PENDING_ACK` state, broker will log a state change to cursor ledger. This ensures the `PENDING_ACK` state is durably stored and can be recovered upon broker failures.
+3. PENDING_ACK -\> PENDING: on receiving [CommandEndTxnOnSubscription](#commandendtxnonsubscription) request with `ABORT` action, the broker logs a state change into cursor ledger of message state transitioning from `PENDING_ACK` to `PENDING`. This ensures the `PENDING_ACK` state can be cleared and the message can be re-dispatched in next consume request.
+4. PENDING_ACK -\> DELETED: on receiving [CommandEndTxnOnSubscription](#commandendtxnonsubscription) request with `COMMIT` action, the broker logs a state change into cursor ledger of message state transitioning from `PENDING_ACK` to `DELETED`. This ensures the message can be deleted and will not be delivered again.
-#### Negative Acknowledgement {#negative-acknowledgement}
+#### Negative Acknowledgement <a id="negative-acknowledgement"></a>
-On broker receiving a negative-ack request, since broker only negative ack messages in \`PENDING\`, broker doesn’t have to do anything.
+On broker receiving a negative-ack request, since broker only negative ack messages in `PENDING`, broker doesn’t have to do anything.
-### Client-Broker Request Processing {#client-broker-request-processing}
+### Client-Broker Request Processing <a id="client-broker-request-processing"></a>
The request processing flow is described as part of state machine above.
-# Future Worker {#future-worker}
+# Future Worker <a id="future-worker"></a>
-## Single Partition Transaction {#single-partition-transaction}
+## Single Partition Transaction <a id="single-partition-transaction"></a>
-Since all the transaction logic are managed via \`Transaction\` instance, there are many optimizations can be done at the client side. For example, we can delay the request of \`CommandNewTxn\` until the transaction is touching \>=2 partitions or \>=1 subscriptions. This allows us falling back to use idempotent producer to achieve transaction on one single partition.
+Since all the transaction logic are managed via `Transaction` instance, there are many optimizations can be done at the client side. For example, we can delay the request of `CommandNewTxn` until the transaction is touching \>=2 partitions or \>=1 subscriptions. This allows us falling back to use idempotent producer to achieve transaction on one single partition.
[^1]: If a global state is involved, the transaction should also consider transactional semantic on updating global state. This proposal will not involve this part as it can be extended and added in the future if we have setup a reliable transaction coordinator.
@@ -1335,20 +1123,12 @@ Since all the transaction logic are managed via \`Transaction\` instance, there
[^3]: We are using ‘partition’ extensively in the remaining of the proposal. A partition is either a non-partitioned topic or a partition in a partitioned topic.
-[^4]: We might consider having a separate TransactionalMessageBuilder to remove \`sendAsync\` and change \`send\` to return Void. because sending a message within a transaction will not complete until the whole transaction is committed or aborted.
+[^4]: We might consider having a separate TransactionalMessageBuilder to remove `sendAsync` and change `send` to return Void. because sending a message within a transaction will not complete until the whole transaction is committed or aborted.
[^5]: We can enable compaction on the managed ledger or take snapshots, or use the bookkeeper state library.
[^6]: Separating Transaction Manager from Message Deduplication is for providing a cleaner separation between normal message flow and transactional message flow.
-[^7]: It is totally fine to drop an \`ACK\`, as there is no guarantee on \`ACK\` before introducing transactional acknowledgement.
-
-[^8]: PENDING\_ACK approach is a pessimistic locking mechanism to lock-in a message until it is committed or aborted. This is totally fine in a streaming application, since a message is anyway delivered to a consumer and most likely consumed by the same consumer. It is very rare that a message is delivered to one consumer but acknowledged by another consumer. So during a life cycle of a message, the consumer is actually acting as a single owner for that message. We can always try the op [...]
-
-[image1]: <data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAnAAAAHtCAMAAACu3bORAAADAFBMVEXl5eW2trbMzMydnZ7p6elgX19ZWFb///+y1vBmZmZCQkJ0dXaIi4xMTEzz8/TX19ff39+QprWkwtcAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA [...]
-
-[image2]: <data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAnAAAAHzCAMAAACXAdB6AAADAFBMVEX///8AAAD09PT7+/vf39/q6upubm5ERESBgYFZWVmlpaXExMSTk5MYGBguLi7S0tK1tbXV1dXB5PeUr72ox9hDT1Z8k5+42uxhc3xTYmqx0eKeu8siKSyIoa8SFRe94PMzPEFvg452jJjH6Ky+3aSuypaNpHqjv40SFRByhWNFUDy21J2YsoRkdVfD5KlVZEojKR+Alm81PS16j2r50t7kwMvZt8K/oaruyNR9anBWSU3NrLYXExRrWl+wlZ1CODssJSj1ztqhh4+PeX+agYjl5eWBmKaFm3OmjJXIyMjb29uwsLCUlJRzc3M9PT0pKSmioqJPT0+EhIQVFRXh4eFiYmK8vLwAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA [...]
-
-[image3]: <data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAnAAAANwCAMAAACYuqclAAADAFBMVEX///8AAAD7+/v09PTf39/ExMRERESlpaWTk5MYGBhZWVkuLi7q6urS0tK1tbWBgYFubm7B5Pe42ux8k5+94PMSFReUr70iKSyeu8tvg44zPEFhc3xTYmpDT1aox9iIoa+x0eLH6Ky+3aSuypaNpHqjv40SFRByhWNFUDy21J2YsoRkdVfD5KlVZEojKR+Alm81PS16j2r50t7uyNTZt8K/oarkwMvNrLZWSU2hh49CODssJSh9anD1ztoXExRrWl+PeX+wlZ1SRUl2jJiBmKaFm3OmjJXl5eXIyMjb29uwsLCUlJRzc3M9PT0pKSmioqJPT0+EhIQVFRXh4eFiYmK8vLwAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA [...]
+[^7]: It is totally fine to drop an `ACK`, as there is no guarantee on `ACK` before introducing transactional acknowledgement.
-[image4]: <data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAmwAAAJYCAMAAADooEqsAAADAFBMVEX////q6uq1tbVZWVkYGBgAAADExMQuLi7f399ERESBgYGlpaXS0tL7+/v09PTb29tubm6Tk5MMfLqp0OZgqdLg7vbE0tr0+fzq9Pnm5uYAAQIDIzVCmcmbyeL7/P1NTU2QkJChwtSUvNF2rszb3d43k8a42OoiiMDG4O+Yx+HU5/J0tNeGvtxMnsxVo85XV1fp6ekbhL5fX185OTl+fn7F3+65zdjN2uFcp9HW1tYAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA [...]
+[^8]: PENDING_ACK approach is a pessimistic locking mechanism to lock-in a message until it is committed or aborted. This is totally fine in a streaming application, since a message is anyway delivered to a consumer and most likely consumed by the same consumer. It is very rare that a message is delivered to one consumer but acknowledged by another consumer. So during a life cycle of a message, the consumer is actually acting as a single owner for that message. We can always try the opt [...]
diff --git a/pip/pip-39.md b/pip/pip-39.md
index cfa7fac35e9..8c5ffe506d1 100644
--- a/pip/pip-39.md
+++ b/pip/pip-39.md
@@ -29,7 +29,7 @@ Figure 1 illustrates the overview of how to use namespace change events to imple
3. Each broker that owns a namespace bundle(s) will subscribe to the \`\_\_change\_events\` topic to receive change events of the namespace. It then appliesapply the change events to the policy cache. (as shown in step 2 “Receive Change Events” and step 3 “Apply Change Events to Policy Cache”)
4. Once the policy cache is updated, the broker sends the response back to the pulsar admin clients. (as shown in step 4 “Http Response”)
-![][image1]
+
Figure 1\. Namespace Change Events & Topic Policy
#### Change Events
@@ -83,5 +83,3 @@ By default infinite retention and compaction are enabled for \`\_\_change\_event
We can still use zookeeper for storing the topic-level policy. See the existing implementation at [https://github.com/apache/pulsar/pull/2659](https://github.com/apache/pulsar/pull/2659)
As explained at the \*Motivation\* section, there are a lot of shortcomings in this approach. For example, more zookeeper accesses when a topic level policy is not set, more zookeeper watches registered if there are many topic level policies are set.
-
-[image1]: <data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAnAAAAGICAYAAADBOt1KAABK/klEQVR4Xu3dB7gU5dn/8aCiJr4aY2wxmggm75vEQv766qsJRYMFsYD0djhlC2BX7CKWJNbYNZYoRuw9atRo7L1X7CCgiB1EIk1k/vc9zLM+5z57DmfP9tnv57qea2bumZ3ds2fm2d/O7s5873sAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAKklDQ8Pfhg4dGtBotI63urq6B+2+BQBAUQwbNiwAUBgnnXRSYPcxAAAK7quvvrKvQQA66JtvviHAAQCKb/78+fY1CEAHEeAAACVBgAMKhwAHACgJAhxQOAQ4AEBJEOCAwiHAAQBKggAHFA4BDgBQEgQ4oHAIcACAkiDAAYVDgAMAlAQBDigcAhwAoCQIcED [...]
\ No newline at end of file
diff --git a/pip/pip-40.md b/pip/pip-40.md
index 125018294df..f6103a8707e 100644
--- a/pip/pip-40.md
+++ b/pip/pip-40.md
@@ -85,7 +85,7 @@ Github Repo*: \[[https://github.com/streamnative/pulsar-manager](https://github.
- *Detailed stats of producers*
- *Detailed stats of consumers*
-![][image2]
+
*The figure above illustrates how does Pulsar Manager work and how it manages multiple Pulsar environments. The Pulsar Manager comprises two parts: frontend and backend.*
@@ -158,31 +158,31 @@ Github Repo*: \[[https://github.com/streamnative/pulsar-manager](https://github.
*\#\#\#\# Dynamic environments with multiple changes*
-*![][image3]*
+**
*\#\#\#\# Tenants management*
-*![][image4]*
+**
*\#\#\#\# Namespaces management*
-*![][image5]*
+**
*\#\#\#\# Topics management*
-*![][image6]*
+**
*\#\#\#\# Subscriptions management*
-*![][image7]*
+**
*\#\#\#\# Brokers management*
-*![][image8]*
+**
*\#\#\#\# Clusters management*
-![][image9]
+
*\#\#\# Test Plan*
@@ -196,19 +196,3 @@ Github Repo*: \[[https://github.com/streamnative/pulsar-manager](https://github.
*If there are alternative ways of accomplishing the same thing, what are they? The purpose of this section is to motivate why a new project is the way it is and not some other way.*
[image1]: <data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAm0AAAALCAYAAAAtObrQAAAAw0lEQVR4Xu3WMY0CABQFQWSdA1QggBIvOEAJCq4+BZgggfz20d4j+ckUo2H38Hu/vui5Pf7gw/PyA6udT0fgyw45GfyvjDWMDCBskzEB+kxbWcYaRgYQtsmYAH2mrSxjDSMDCNtkTIA+01aWsYaRAYRtMiZAn2kry1jDyADCNhkToM+0lWWsYWQAYZuMCdBn2soy1jAygLBNxgToM21lGWsYGUDYJmMC9Jm2sow1jAwgbJMxAfpMW1nGGkYGELbJmAB9bxL59nuswkk6AAAAAElFTkSuQmCC>
-
-[image2]: <data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAnAAAAHzCAYAAACg3yBIAACAAElEQVR4Xuydh5cbRfa2v78J+xB/pCVnll3iLjktGZZlgSUnk3P04ICNjRNOOGBsbBOcbXlmnHNOGEdsHGemvn5KKrmnutVq9ZRGLfW957xnNK1uqav6quqtm+r/KRERERERERERkbqS/2cfEBERERERERERSbcIgRMRERERERERqTMRAiciIiIiIiIiUmciBE5EREREREREpM5ECJyIiIiIiIiISJ2JEDgRERERERERkToTIXAiIiIiIiIiInUmQuBEREREREREROpMQglcx5b9qmPVLtWx+DeBQCBIN5b9rjo27rOHsdLS3i5jnEAgqB94Y5zae9geyUIIXJs3uOW2e/hNIBAI6gQ7lNp9yB7NQkUTNxnjBAJBXWGbsse4AIHr2HbAOxG0eegQCASClINF51G [...]
-
-[image3]: <data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAnAAAAFfCAMAAAD53wHvAAADAFBMVEUAAAAhIyUtLjEvMDMyMzU9PkA/QEIzWX9ISUxMTlBOUFJWV1lBX39dXmBGY39eYmdVan9jZGZjZWlmaGxqa2xsbnBtcHR4eHo8XYA/YIBaFoxlE5hhH5JiGZltEqBsHKlzE6d1F7F6FLV+Frl4GbJiJJJrJ5t1J6l9KbNGYoBWa4BfcIBgb4Boc4B4fIAUw5ogxp8ixqA1y6g8zat9gIJAnv9Jov9Wqf9hrv9psv95uv9P0rNS07Rd1rlh1rpm2L1q2b923MR63cbzbR7ybiLzcCTzdSz4cyb0ezX1bGz1dHSPA5OVB5iWE5iEE7eWDaaVG6KSE72BJriAKbaYJ6OWJL+bMKanBrmsGaWsFre2Fa+xF7imJqaiNainMbe+L7C2Pa61N7aUFc2VGsudE8W [...]
-
-[image4]: <data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAnAAAAFfCAMAAAD53wHvAAADAFBMVEUAAAAaGhoXISweJzIiIyUpLTErMTcyMzQbPlwtN0A4PEIeQF0iQl48QEUpR2E0TmU6Umd7Njd8P0BISUtMTVFOUVRVVlhJVmRWW2NPYG5dYmdWZXF8SEh9Vld+X2BiY2VjZWpmaGtqa21qbXFtcHN+Z2h4eHogT4AmUoA3W4BaFoxlE5hhH5JiGZltFqRzE6d1F7F6FLV+Frl4GbJmJZZ1J6l9KbNAXoBIY4BWaoBfcIBnc4B5fIAUw5ogxp8ixqA7zap+gIJAnv9Iov9WqP9hrv9psv95uv9P0rNT07Vj17t23MR63caAe3vzbR7ybiLzcCTzdSz4cyb0ezX1bGz1dHSPA5OVB5iWE5iEE7eWDaaVG6KSE72BJriAKbaYJ6OWJL+bMKanBrmsGaWsFre [...]
-
-[image5]: <data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAnAAAAFfCAMAAAD53wHvAAADAFBMVEUAAAAYGRoeHiAXICseJzIlJictLjAsMDQzMzUaPVwtN0A6PUEeQF0hQV49QUYpR2EzTWQ6Umd7Njd8P0BHSEpMTVBOUVNWV1hKVmNVW2NOYG5dYmdYZnF8RkZ9WVl+X2BlZmhrbXFtcHN+aWl3d3kgT4AmUoA3W4BaFoxlE5hhH5JiGZltFqRzE6d1F7F6FLV+Frl4GbJmJZZ1J6l9KbNAXoBIY4BWaoBfcIBgb4BodIB5fIAUw5ogxp8ixqA7zap+gIFAnv9Iov9WqP9hrv9psv95uv9P0rNT07Vj17t23MR63caAenrzbR7ybiLzcCTzdSz4cyb0ezX1bGz1dHSPA5OVB5iWE5iEE7eWDaaVG6KSE72BJriAKbaYJ6OWJL+bMKanBrmsGaWsFre2Fa+ [...]
-
-[image6]: <data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAnAAAAFfCAMAAAD53wHvAAADAFBMVEUAAAAYGRoeHiAXICseJzImJygsLjAsMDQ0NTcbPVwtN0A8PkAeQF0iQl4+QEIpR2EyTWQ6UmdHSEpMTlBOUFJWVlhIV2ZZXGJPYG5cYWhZZnJiY2ViZmplaGxra2xpbXJtcHN3eHkgT4AmUoA3WoBaFoxlE5hhH5JiGZltFqRzE6d1F7F6FLV+Frl4GbJmJZZ1J6l9KbNAX4BIY4BXa4BfcIBgb4Boc4B6fYAUw5ogxp8ixqA7zap+gIFAnv9Iov9Yqf9hrv9osv92uf9P0rNT07Vj17t23MR63cbzbR7ybiLzcCTzdSz4cyb0ezX1bGz1dHSPA5OVB5iWE5iEE7eWDaaVG6KSE72BJriAKbaYJ6OWJL+bMKanBrmsGaWsFre2Fa+xF7imJqaiNainMbe [...]
-
-[image7]: <data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAnAAAAFfCAMAAAD53wHvAAADAFBMVEUAAAAZGRkaHSAiIyYtLjAvMDM0NTcnNkY9PkAeQWQvQVY+QEI0Rlo8TmJGR0hNTlBOUFJWV1hBU2ZeXmBeYmdUZXhgYmVjZWlmaGxpam1rbXFucHR0dnlaFoxlE5hhH5JiGZltFqRzE6d1F7F6FLV+Frl4GbJmJZZ1J6l9KbNefJpldYd8foE/nf0Uw5ogxp8ixqA1y6g8zat+gIN3hZdSjcpVkdBAnv9RnvBIov5Xqf9tpeBhrv9osv92uf9P0rNT07Vj17t23MR63cbzbR7ybiLzcCTzdSz4cyb0ezWPA5OVB5iWE5iEE7eWDaaVG6KSE72BJriAKbaYJ6OWJL+bMKanBrmsGaWsFre2Fa+xF7imJqaiNainMbe+L7C2Pa61N7aZFceRKMWmCcimCtK [...]
-
-[image8]: <data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAnAAAAFfCAMAAAD53wHvAAADAFBMVEUAAAAiIyYtLjEvMDMyMzQ9PkAvQVY/QEI2R1w8TmJGR0hNTlBOUFFWV1hBU2ZdXmBeYmdUZXhiZGdrbXFucHR1dng0fLo0fsFaFoxlE5hhH5JiGZltFqRzE6d1F7F6FLV+Frl4GbJmJZZ1J6l9KbNldYd8foE+gLo2gsg4idU6juA8kuk+mfYUw5ogxp8ixqA7zapGhr1+gIN3hZdNisBSjsFZk8RAnf5Jov9Wqf9nm8ltoMt0pM57qdBhrv9osf93uv9P0rNT07Vj17t23MR63cbzbR7ybiLzcCTzdSz4cyb0ezWPA5OVB5iWE5iEE7eWDaaVG6KSE72BJriAKbaYJ6OWJL+bMKanBrmsGaWsFre2Fa+xF7imJqaiNainMbe+L7C2Pa61N7aZFceRKMW [...]
-
-[image9]: <data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAnAAAAFfCAMAAAD53wHvAAADAFBMVEUAAAAiIyYtLjEvMDMyMzU9PkAvQVY/QEI2R1w8TmJGR0hNTlBOUFFWV1hBU2ZdXmBeYmdUZXhgYmVjZWlmaGxpam1rbXFucHR1dnlaFoxlE5hhH5JiGZltFqRzE6d1F7F6FLV+Frl4GbJmJZZ1J6l9KbNldYd8fYEUw5ogxp8ixqA7zap+gIN3hZdAnv9Iov9Yqv9hrv9psv94uv9P0rNT07Vj17t23MR63cb/AADzbR7ybiLzcCTzdSz4cyb0ezXldHD1bGz1dnaPA5OVB5iWE5iEE7eWDaaVG6KSE72BJriAKbaYJ6OWJL+bMKanBrmsGaWsFre2Fa+xF7imJqaiNainMbe+L7C2Pa61N7aZFceRKMWmCcimC9aqF8imFtSgGNOiHdiqEdOsFd6rG9O [...]