[fix][broker] Fix potential NPE when disabling the liveness check

[mattisonchao]

Motivation

Fix potential NPE when disabling the liveness check

Modifications

• Fix potential NPE when disabling the liveness check

Verifying this change

• Make sure that the change passes the CI checks.

Documentation

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

[github-actions]

@mattisonchao 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 -->

[RobertIndie]

LGTM.

Just want to make sure. The NPE will only be thrown here:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java

Line 971 in c4cff0a

if (previousIsActive) {

Right?

[lhotari]

I think that there was a reason to return null when this was implemented. Please postpone merging until I have reviewed this.

[lhotari]

@mattisonchao is there a way to have a test case for this?

[Technoboy-]

@mattisonchao is there a way to have a test case for this?

I don't think we need this kind of test, the method should return Boolean, not null.

[dao-jun]

LGTM, but also need to remove the null checker?
https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java#L173

[lhotari]

I now had a chance to review the reason why I had originally used null.
It's documented here:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TransportCnx.java

Lines 80 to 87 in c4cff0a

	/***
	* Check if the connection is still alive
	* by actively sending a Ping message to the client.
	*
	* @return a completable future where the result is true if the connection is alive, false otherwise. The result
	* is null if the connection liveness check is disabled.
	*/
	CompletableFuture<Boolean> checkConnectionLiveness();

I'll suggest that we change the return value from CompletableFuture<Boolean> to CompletableFuture<Optional<Boolean>> if there's a desire to get rid of the special null usage.
@mattisonchao Could you make this change?

[Technoboy-]

I now had a chance to review the reason why I had originally used null. It's documented here:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TransportCnx.java

Lines 80 to 87 in c4cff0a

	/***
	* Check if the connection is still alive
	* by actively sending a Ping message to the client.
	*
	* @return a completable future where the result is true if the connection is alive, false otherwise. The result
	* is null if the connection liveness check is disabled.
	*/
	CompletableFuture<Boolean> checkConnectionLiveness();

I'll suggest that we change the return value from CompletableFuture<Boolean> to CompletableFuture<Optional<Boolean>> if there's a desire to get rid of the special null usage. @mattisonchao Could you make this change?

ok, then we find a new place need to fix to avoid NPE

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java

Lines 958 to 979 in cea5c93

	private CompletableFuture<Void> tryOverwriteOldProducer(Producer oldProducer, Producer newProducer) {
	if (newProducer.isSuccessorTo(oldProducer)) {
	oldProducer.close(false);
	if (!producers.replace(newProducer.getProducerName(), oldProducer, newProducer)) {
	// Met concurrent update, throw exception here so that client can try reconnect later.
	return CompletableFuture.failedFuture(new BrokerServiceException.NamingException("Producer with name '"
	+ newProducer.getProducerName() + "' replace concurrency error"));
	} else {
	handleProducerRemoved(oldProducer);
	return CompletableFuture.completedFuture(null);
	}
	} else {
	// If a producer with the same name tries to use a new connection, async check the old connection is
	// available. The producers related the connection that not available are automatically cleaned up.
	if (!Objects.equals(oldProducer.getCnx(), newProducer.getCnx())) {
	return oldProducer.getCnx().checkConnectionLiveness().thenCompose(previousIsActive -> {
	if (previousIsActive) {
	return CompletableFuture.failedFuture(new BrokerServiceException.NamingException(
	"Producer with name '" + newProducer.getProducerName()
	+ "' is already connected to topic"));
	} else {
	// If the connection of the previous producer is not active, the method

[codecov-commenter]

Codecov Report

All modified and coverable lines are covered by tests ✅

Comparison is base (1a4a9d9) 73.66% compared to head (278372e) 73.60%.
Report is 10 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #21840      +/-   ##
============================================
- Coverage     73.66%   73.60%   -0.07%     
+ Complexity    32343    32321      -22     
============================================
  Files          1858     1858              
  Lines        138146   138174      +28     
  Branches      15141    15148       +7     
============================================
- Hits         101763   101699      -64     
- Misses        28544    28606      +62     
- Partials       7839     7869      +30     

Flag	Coverage Δ
inttests	24.20% <33.33%> (+0.01%)	⬆️
systests	23.70% <0.00%> (-0.17%)	⬇️
unittests	72.89% <100.00%> (-0.06%)	⬇️

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

Files	Coverage Δ
...ervice/AbstractDispatcherSingleActiveConsumer.java	90.55% <100.00%> (ø)
...rg/apache/pulsar/broker/service/AbstractTopic.java	87.71% <100.00%> (-0.21%)	⬇️
...va/org/apache/pulsar/broker/service/ServerCnx.java	72.30% <100.00%> (-0.06%)	⬇️

... and 69 files with indirect coverage changes

[mattisonchao]

Hi, @lhotari @Technoboy-

      * @return a completable future where the result is true if the connection is alive, false otherwise. The result 
      * is null if the connection liveness check is disabled. 

1. I found no particular usage for the value null. I am trying to understand why we should have a third status here.
2. In my opinion, if we turned off the connection liveness check, that logic should be equal to the connection always being alive.

[lhotari]

2. In my opinion, if we turned off the connection liveness check, that logic should be equal to the connection always being alive.

@mattisonchao I disagree. That would be bad API design. There might be future use cases where this matters.