[fix] [broker] Fix isolated group not work problem.

[horizonzy]

Modifications

When upgraded the pulsar version from 2.9.2 to 2.10.3, and the isolated group feature not work anymore.
Finally, we found the problem. In IsolatedBookieEnsemblePlacementPolicy, when it gets the bookie rack from the metadata store cache, uses future.isDone() to avoid sync operation. If the future is incomplete, return empty blacklists.
The cache may expire due to the caffeine cache getExpireAfterWriteMillis config, if the cache expires, the future may be incomplete. (#21095 will correct the behavior)

In 2.9.2, it uses the sync to get data from the metadata store, we should also keep the behavior.

Documentation

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

Matching PR in forked repository

PR in forked repository:

[hangc0276]

Please add a test to protect this logic.

[codecov-commenter]

Codecov Report

Merging #21096 (c033ac0) into master (b26ee8a) will increase coverage by 39.72%.
The diff coverage is 64.28%.

@@              Coverage Diff              @@
##             master   #21096       +/-   ##
=============================================
+ Coverage     33.46%   73.18%   +39.72%     
- Complexity    12179    32436    +20257     
=============================================
  Files          1623     1887      +264     
  Lines        127399   139985    +12586     
  Branches      13929    15413     +1484     
=============================================
+ Hits          42631   102449    +59818     
+ Misses        79158    29440    -49718     
- Partials       5610     8096     +2486     

Flag	Coverage Δ
inttests	24.14% <0.00%> (+0.06%)	⬆️
systests	25.11% <0.00%> (?)
unittests	72.47% <64.28%> (+40.52%)	⬆️

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

Files Changed	Coverage Δ
...areness/IsolatedBookieEnsemblePlacementPolicy.java	87.78% <64.28%> (+87.78%)	⬆️

... and 1532 files with indirect coverage changes

[Technoboy-]

LGTM

[hangc0276]

Nice catch!

[codelipenghui]

Can the zookeeper thread reach here? Just to confirm it will not introduce a deadlock.

[horizonzy]

I discuss with hang, and we will use a graceful way to avoid sync operation. I will push a next commit later.

[hangc0276]

LGTM. Please take a look at the failed tests, thanks.

[eolivelli]

+1
Great catch

[codelipenghui]

I would like to provide another approach for handling the cached rack configuration update.

1. Make allGroupsBookieMapping as a volatile variable
2. Update the cached rack configuration async

initialize

bookieMappingCache.get(BookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH).thenAccept(opt -> {
                opt.ifPresent(bookiesRackConfiguration -> cachedRackConfiguration = bookiesRackConfiguration);
            }).exceptionally(ex -> {
                log.warn("Failed to load bookies rack configuration while initialize the PlacementPolicy.");
                return null;
            });

getExcludedBookiesWithIsolationGroups

                CompletableFuture<Optional<BookiesRackConfiguration>> future =
                        bookieMappingCache.get(BookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH);

                BookiesRackConfiguration allGroupsBookieMapping;
                future.thenAccept(opt ->
                        opt.ifPresent(bookiesRackConfiguration -> cachedRackConfiguration = bookiesRackConfiguration)
                    ).exceptionally(ex -> {
                        log.warn("The newest bookie rack config is not available now.");
                        return null;
                    });
                allGroupsBookieMapping = cachedRackConfiguration;
                if (allGroupsBookieMapping == null) {
                    return excludedBookies;
                }

When the broker restarts, we might lose the rack configurations, but it can avoid any potential deadlocks.

The initialize method is called when starting the broker before the broker service becomes available to the client. So we will not lose the bookie rack configuration in most cases.

[codelipenghui]

How bookie client will handle this exception, and what is the expected behavior from the producer/consumer's perspective? Will the producer/consumer be closed by this exception?

[horizonzy]

Not, we already caught this exception at line_284

[horizonzy]

I would like to provide another approach for handling the cached rack configuration update.

1. Make allGroupsBookieMapping as a volatile variable
2. Update the cached rack configuration async

initialize

bookieMappingCache.get(BookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH).thenAccept(opt -> {
                opt.ifPresent(bookiesRackConfiguration -> cachedRackConfiguration = bookiesRackConfiguration);
            }).exceptionally(ex -> {
                log.warn("Failed to load bookies rack configuration while initialize the PlacementPolicy.");
                return null;
            });

getExcludedBookiesWithIsolationGroups

                CompletableFuture<Optional<BookiesRackConfiguration>> future =
                        bookieMappingCache.get(BookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH);

                BookiesRackConfiguration allGroupsBookieMapping;
                future.thenAccept(opt ->
                        opt.ifPresent(bookiesRackConfiguration -> cachedRackConfiguration = bookiesRackConfiguration)
                    ).exceptionally(ex -> {
                        log.warn("The newest bookie rack config is not available now.");
                        return null;
                    });
                allGroupsBookieMapping = cachedRackConfiguration;
                if (allGroupsBookieMapping == null) {
                    return excludedBookies;
                }

When the broker restarts, we might lose the rack configurations, but it can avoid any potential deadlocks.

The initialize method is called when starting the broker before the broker service becomes available to the client. So we will not lose the bookie rack configuration in most cases.

It makes sense, we should print a debug log when the allGroupsBookieMapping is null.

[codelipenghui]

It makes sense, we should print a debug log when the allGroupsBookieMapping is null.

Yes, sure.

[horizonzy]

It makes sense, we should print a debug log when the allGroupsBookieMapping is null.

Yes, sure.

There is a case that needs discussed, we only update the rack config when the opt.isPresent. If the user deletes the zk nodes of path /bookies, the opt.isPresent will be false. Shall we delete the cached rack config or still use it?

[hangc0276]

It makes sense, we should print a debug log when the allGroupsBookieMapping is null.

Yes, sure.

There is a case that needs discussed, we only update the rack config when the opt.isPresent. If the user deletes the zk nodes of path /bookies, the opt.isPresent will be false. Shall we delete the cached rack config or still use it?

@horizonzy We need to delete the cached rack configuration.

[horizonzy]

It makes sense, we should print a debug log when the allGroupsBookieMapping is null.

Yes, sure.

There is a case that needs discussed, we only update the rack config when the opt.isPresent. If the user deletes the zk nodes of path /bookies, the opt.isPresent will be false. Shall we delete the cached rack config or still use it?

@horizonzy We need to delete the cached rack configuration.

What do you think of it?

[hangc0276]

It makes sense, we should print a debug log when the allGroupsBookieMapping is null.

Yes, sure.

There is a case that needs discussed, we only update the rack config when the opt.isPresent. If the user deletes the zk nodes of path /bookies, the opt.isPresent will be false. Shall we delete the cached rack config or still use it?

@horizonzy We need to delete the cached rack configuration.

What do you think of it?

@horizonzy Do checking the exception type work? If the exception is a NoNode exception, update the cached rack configuration to null.

[horizonzy]

It makes sense, we should print a debug log when the allGroupsBookieMapping is null.

Yes, sure.

There is a case that needs discussed, we only update the rack config when the opt.isPresent. If the user deletes the zk nodes of path /bookies, the opt.isPresent will be false. Shall we delete the cached rack config or still use it?

@horizonzy We need to delete the cached rack configuration.

What do you think of it?

@horizonzy Do checking the exception type work? If the exception is a NoNode exception, update the cached rack configuration to null.

It doesn't work. The zk metadata store handles it separately, not throw exceptions.

pulsar/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/ZKMetadataStore.java

Lines 267 to 269 in b26ee8a

	if (code == Code.NONODE) {
	// For get operations, we return an empty optional
	op.getFuture().complete(Optional.empty());

[hangc0276]

Change the log msg to log.warn("Failed to update the newest bookies rack config, and use the cached rack configuration: {}", cachedRackConfiguration)

[horizonzy]

This is an async operation, it won't affect the result of the current invocation, and printing this log may be misleading.

[hangc0276]

make sense

[codelipenghui]

/pulsarbot run-failure-checks

[shibd]

@horizonzy Can you help create a PR to cherry-pick this change to branch-2.11?

[horizonzy]

@horizonzy Can you help create a PR to cherry-pick this change to branch-2.11?

Ok.