[improve][broker] PIP-192 Made only the leader consume TopBundlesLoadDataStore

[heesung-sn]

Master Issue: #16691

Motivation

Raising a PR to implement #16691.

Based on the current design, only the leader needs to consume the topk bundle load data.

Modifications

This PR

• made only the leader consume TopBundlesLoadDataStore
• moved LeaderElectionService to ExtensibleLoadManager from ServiceUnitStateChannel.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• *Updated unit tests.

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• Anything that affects deployment

Documentation

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

We will have separate PRs to update the Doc later.

Matching PR in forked repository

PR in forked repository: heesung-sn#37

[Demogorgon314]

@heesung-sn Please help rebase PR to the master branch.

[heesung-sn]

@heesung-sn Please help rebase PR to the master branch.

Done.

[Demogorgon314]

After being interrupted, do we still need to start the table view?

[heesung-sn]

Optimistically, yes.

If the load store has been initialized, then it will successfully start table view.

If not, in the worst case, the table view start will throw an exception.

[heesung-sn]

In fact, I think we can retry when interrupted.

[gaoran10]

For TopKBundle tableView, it seems that if the check validateTableViewStart is failed, the UnloadScheduler will also fail to find bundles that are needed to unload, can we try to re-initialize the tableView in the method validateTableViewStart?

[heesung-sn]

When ExtensibleLoadManager.start() and playLeader() are complete, then topBundlesLoadStore.validateTableViewStart() should never fail from the leader broker(from UnloadScheduler).

Currently, follower brokers should never access topBundlesLoadDataStore.tableview, and if they do, we should throw an exception in this case instead of re-initializing the tableView(followers don't run UnloadScheduler).

I also moved the UnloadScheduler.start()/close() logic into the playLeader() and playFollower().

[gaoran10]

According to this code block, even if it failed to start the topBundlesLoadDataStore, the playLeader still can complete, if the UnloadScheduler tries to find bundles that need to unload, it will try to get topKBundlesData, at this moment it will encounter IllegalStateException("table view has not been started").

    private void playLeader() {
        this.pulsar.getLoadManagerExecutor().execute(() -> {
            serviceUnitStateChannel.scheduleOwnershipMonitor();
            waitForLoadStoreInit();
            try {
                topBundlesLoadDataStore.startTableView();
            } catch (Throwable e) {
                log.error("The new leader:{} failed to start topBundlesLoadDataStore tableview",
                        pulsar.getLookupServiceAddress(), e);
            }
            ...
            log.info("This broker:{} is the leader now.", pulsar.getLookupServiceAddress());
        });
    }

TransferShedder.java

public UnloadDecision findBundlesForUnloading(LoadManagerContext context,
                                                  Map<String, Long> recentlyUnloadedBundles,
                                                  Map<String, Long> recentlyUnloadedBrokers) {
    try {
        ...
        // If topBundlesLoadDataStore is not initialized, IllegalStateException will be thrown.
        Optional<TopBundlesLoadData> bundlesLoadData = context.topBundleLoadDataStore().get(maxBroker);
        ...
    }  catch (Throwable e) {
        log.error("Failed to process unloading. ", e);
        decision.fail();
    }
    return decision;

}

[heesung-sn]

ExtensibleLoadManager.start() and playLeader() are complete

I meant this complete as everything initialized correctly without an error.

Yes. Currently, if topBundlesLoadDataStore.startTableView(); fails, the leader broker still proceeds, and the TransferShedder will constantly make UnloadDecision(fail).

However, lazy-init(tableview.start ) in validateTableViewStart could also fail and lead to the same situation (TransferShedder will constantly make UnloadDecision(fail)). Besides, this lazy-init can lead to unknown behaviors(many consumers on this topBundlesLoad topic) when other brokers access the tableview.

The bigger question is how Pulsar handles system state change failures. In this case, the state change is from playLeader or playFollower.

I can think of the following options:

Option 1: retry-forever
Option 2: fail-fast(shutdown broker)
Option 3: ignore (leader or followers will run in the invalid state, and there will be many error logs and metrics. Eventually, the operator needs to capture this and fix it, probably by broker restart)

AFAIK, upon system state change failures, other systems immediately fail fast with meaningful logs and even with coredump. However, I don't think this fail-fast is the practice in the Pulsar project.

I can update the code for option 1 (retry-forever).

Please let me know if you have more concerns.

[gaoran10]

Thanks for your explanation.

Obviously, if everything is initialized correctly without an error, it's OK, but you caught exceptions while starting topBundlesLoadDataStore before.

However, lazy-init(tableview.start ) in validateTableViewStart could also fail and lead to the same situation (TransferShedder will constantly make UnloadDecision(fail))

Yes, the lazy initialization in validateTableViewStart could also fail, but it will not fail forever. If we only caught the exception, it will fail forever.

I can update the code for option 1 (retry-forever).

I think it's ok, we can merge it first.

[Demogorgon314]

@heesung-sn Please help rebase this PR to the master branch to resolve the conflicts.

[heesung-sn]

@heesung-sn Please help rebase this PR to the master branch to resolve the conflicts.

Lets merge the other PRs and rebase this PR again.

[heesung-sn]

@heesung-sn Please help rebase this PR to the master branch to resolve the conflicts.

Lets merge the other PRs and rebase this PR again.

@Demogorgon314 rebase is done.

[heesung-sn]

@Demogorgon314 Please merge this PR if looking good.

[Technoboy-]

/pulsarbot rerun-failure-checks

[heesung-sn]

/pulsarbot rerun-failure-checks