[improve] Introduce the sync() API to ensure consistency on reads during critical metadata operation paths

[eolivelli]

This is only a POC, opened for discussion

Explanation

When a operation is redirected to another broker, it may happen that the local view over metadata on the other broker is not up-to-date with the view seen by the initial broker.
In order to guarantee casual consistency among different peers you have to ways using the ZooKeeper APIs:

• perform a write
• use the sync() API

Both of the two operations force (or wait for) the ZooKeeper server to which the client is connected to be in sync with the Leader ZooKeeper server and also that the local ZooKeeper client is up-to-date.

See more at https://zookeeper.apache.org/doc/r3.8.0/zookeeperProgrammers.html at the "Consistency Guarantees" paragraph.

When the control passes from one broker to another broker the second broker needs to ensure that its local view is not stale.
And we can do it by using the sync() API and then invalidating the local MetadataCache.
The guarantee here is that the operation executed on the initial broker "happened before" the operation on the second broker.

A good example of the need for this feature is described on PR #18193.
at comment https://github.com/apache/pulsar/pull/18193/files#r1005919247

Modifications

Add to MetadataStore the support for using the sync() API and add refreshAndGetAsync() to MetadataCache.
Fix the problem reported by https://github.com/apache/pulsar/pull/18193/files#r1005919247 regarding Partitioned Topic deletion (the same fix should be applied to Namespace deletion, about the Policies#deleted flag)

PR in the forked repository: eolivelli#20

• doc
• doc-required
• doc-not-needed
• doc-complete

[github-actions]

@eolivelli Please add the following content to your PR description and select a checkbox:

- [ ] `doc` <!-- Your PR contains doc changes -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->

[codecov-commenter]

Codecov Report

Merging #18518 (c7b7c31) into master (b42aed1) will increase coverage by 0.09%.
The diff coverage is 41.17%.

@@             Coverage Diff              @@
##             master   #18518      +/-   ##
============================================
+ Coverage     47.11%   47.21%   +0.09%     
- Complexity    10595    10685      +90     
============================================
  Files           710      711       +1     
  Lines         69423    69455      +32     
  Branches       7449     7452       +3     
============================================
+ Hits          32709    32792      +83     
+ Misses        33037    32993      -44     
+ Partials       3677     3670       -7     

Flag	Coverage Δ
unittests	47.21% <41.17%> (+0.09%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
...pulsar/broker/admin/v2/ExtNonPersistentTopics.java	0.00% <0.00%> (ø)
...n/java/org/apache/pulsar/broker/PulsarService.java	56.57% <33.33%> (-1.48%)	⬇️
...pache/pulsar/client/impl/TableViewBuilderImpl.java	72.41% <55.55%> (+72.41%)	⬆️
...a/org/apache/pulsar/client/impl/TableViewImpl.java	29.16% <75.00%> (+29.16%)	⬆️
...he/pulsar/broker/admin/v2/ExtPersistentTopics.java	76.92% <100.00%> (ø)
...pulsar/client/impl/TableViewConfigurationData.java	43.75% <100.00%> (+8.03%)	⬆️
...pache/pulsar/proxy/server/ProxyServiceStarter.java	60.92% <100.00%> (+0.26%)	⬆️
...apache/pulsar/broker/service/TopicListService.java	10.40% <0.00%> (-30.40%)	⬇️
...ar/broker/loadbalance/impl/BundleSplitterTask.java	54.00% <0.00%> (-28.00%)	⬇️
...lsar/broker/loadbalance/impl/ThresholdShedder.java	3.27% <0.00%> (-27.05%)	⬇️
... and 59 more

[merlimat]

Apart from the fact that sync() is a very ZK specific concept, my main problem with this is what is this supposed to achieve and where is this supposed to be used.

Is this to fix the problem of a partitioned topic created and immediately used by a client and seeing error?

The part that is important to highlight is that sync() is essentially forcing a write in the ZK log, in order to guarantee read-after-write in ZK. Imposing a sync each time a client connects, will open a direct write path from clients to ZK.

That would mean that even when using a fully-consistent data store, we would still have the problem that we cannot cache this information.

I would rather find a better way, where we only force the sync on very specific occasions. eg:

1. try to get partitioned-topic-metadata
2. It's not in local cache
3. Try to read from metadata store
4. We get NotFound error
5. Force sync() and redo the read
6. Cache the result of the read, (including the NotFound result)

[poorbarcode]

LGTM. By the way, should we need a test for this?

[poorbarcode]

Hi @merlimat

Is this to fix the problem of a partitioned topic created and immediately used by a client and seeing error?

No, What this PR is trying to fix is that when the metadata attribute deleted of a partitioned topic has changed to true, this value is still cached in other brokers. This PR makes sync and read in other brokers. The first discussion is https://github.com/apache/pulsar/pull/18193/files#r1005921385

I would rather find a better way.

I agree. +1

[poorbarcode]

Should we need a test for this?

[eolivelli]

Thank you @merlimat and @poorbarcode for your comments.

@merlimat this proposal is only to cover very specific cases, as you say we are already doing well for most of the cases.

The case in which you need "sync()" is when there is this combo happening:

• you have 2 brokers
• broker A mutates the metadata (op1)
• another action triggers a "read" (op2) followed by a write (op3) on broker B

In this case when broker B executes op2 we must guarantee that the value that observes includes the effects generated by op1, otherwise op3 may be executed using stale data (any compare-and-set operation in op3 is not on the same znode as op2, so we not protection, no BadVersion).

In the partitioned topic deletion example we have:

• op1 -> PartitionedTopicMetadata.deleted = true on Broker A
• op2 -> Read (from cache or from the ZK follower node in case of cache miss) PartitionedTopicMetadata on Broker B
• op3 -> Do something (like creating again a partition in case of autocreation enabled) on Broker B

You need sync() (on broker B) before op2 because the action that causes op2 is out-of-band in respect to ZooKeeper, and so Broker B may use a stale version of the Z-Node (either because it is connected to a Follower that is not up-to-date or because it has a stale value in the local cache and the watch notification still hasn't arrived, mostly because the Follower is still not up-to-date).

So we have to use this "consistent refreshAndGet" only when this pattern happens.

No need to change Pulsar everywhere, but only on SOME of the Write paths that touch Metadata.

I would really reject any proposal to call sync() before every read as it would kill the existence of the localcache and kill also the performances (as you say, because sync() is a dymmy write)or every write as CAS already covers 99% of the operations

[eolivelli]

@poorbarcode I will add tests only when we reach consensus on the approach.

[eolivelli]

@merlimat do you have more comments please ?

[github-actions]

The pr had no activity for 30 days, mark with Stale label.

[eolivelli]

@merlimat @codelipenghui @hangc0276 @rdhabalia I believe that this PR adds value and that we need to complete this work.

I can't find any other way to achieve the goals of this PR

[poorbarcode]

@eolivelli I want to cherry-pick this PR to 2.10 and 2.11 to solve the issue, which #21063 fixed. Do you feel okay?

[poorbarcode]

@eolivelli I want to cherry-pick this PR to 2.10 and 2.11 to solve the issue, which #21063 fixed. Do you feel okay?

I sent a discuss to cherry-pick this PR to branch-2.10 and branch-2.11

• https://lists.apache.org/thread/g6b1xh3tr8qj0vdd0jwtkhswlsr50631

[poorbarcode]

I sent a discuss to cherry-pick this PR to branch-2.10 and branch-2.11

• https://lists.apache.org/thread/g6b1xh3tr8qj0vdd0jwtkhswlsr50631

Cherry-picked.

Since the method isPartitionedTopicBeingDeletedAsync only exists in branch-3.0, the change https://github.com/apache/pulsar/pull/18518/files#diff-38fac527320b93f0e9ea2459420f56638299627f7f29c01484f7e4aae0c5f0b7R330 was not be cherry-picked into branch-2.10 and branch-2.11