[improve] [broker] Add consumer-id into the log when doing subscribe. #20568

poorbarcode · 2023-06-13T08:45:18Z

Modifications

(important) Since cnx.address + consumerId is the identifier of one consumer; add consumer-id into the log when doing subscribe.
add a test to confirm that even if the error occurs when sending messages to the client, the consumption is still OK.
print debug log if ack-command was discarded due to ConsumerFuture is not complete.
print debug log if sending a message to the client is failed.

Documentation

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

Matching PR in forked repository

PR in forked repository: x

lifepuzzlefun · 2023-06-13T14:01:49Z

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

@@ -361,6 +361,12 @@ public Future<Void> sendMessages(final List<? extends Entry> entries, EntryBatch
                msgOutCounter.add(totalMessages);
                bytesOutCounter.add(totalBytes);
                chunkedMessageRate.recordMultipleEvents(totalChunkedMessages, 0);
+            } else {
+                log.warn("[{}-{}] Sent messages to client fail by IO exception[{}], these messages(messages count:"


the {} in log seems not align to the parameter. IO exception[{}] is the third {}, in the parameter is the last

Ah, you are right. already fixed, Thanks

Is it better to change the log level to debug? It's a message level log, for a cluster with high consumption throughput, it will introduce too many warning logs. We should be careful to add logs on the message level with >= info level.

mattisonchao · 2023-06-15T03:47:26Z

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

@@ -361,6 +361,13 @@ public Future<Void> sendMessages(final List<? extends Entry> entries, EntryBatch
                msgOutCounter.add(totalMessages);
                bytesOutCounter.add(totalBytes);
                chunkedMessageRate.recordMultipleEvents(totalChunkedMessages, 0);
+            } else {
+                log.warn("[{}-{}] Sent messages to client fail by IO exception[{}], these messages(messages count:"


Have you considered the connection problems that will print tons of logs here? :)

IMO, the message trace logs should be considered carefully.

Have you considered the connection problems that will print tons of logs here? :)
IMO, the message trace logs should be considered carefully.

Added a health check here, if the health check fail, this connection will be closed immediately. So this log only be printed when the health check is in progress.

mattisonchao · 2023-06-15T03:49:52Z

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

@@ -1783,6 +1784,10 @@ protected void handleAck(CommandAck ack) {
                }
                return null;
            });
+        } else {
+            log.warn("Consumer future is not complete(not complete or error), but received command ack. so discard"


Same concern as before. Tons of logs?

Already change logger level to DEBUG

codecov-commenter · 2023-06-15T06:29:46Z

Codecov Report

Merging #20568 (f5c17ec) into master (51c2bb4) will increase coverage by 39.50%.
The diff coverage is 89.85%.

@@              Coverage Diff              @@
##             master   #20568       +/-   ##
=============================================
+ Coverage     33.50%   73.01%   +39.50%     
- Complexity    12053    31966    +19913     
=============================================
  Files          1613     1867      +254     
  Lines        126120   138662    +12542     
  Branches      13749    15236     +1487     
=============================================
+ Hits          42254   101239    +58985     
+ Misses        78332    29385    -48947     
- Partials       5534     8038     +2504

Flag	Coverage Δ
inttests	`24.18% <10.14%> (+0.01%)`	⬆️
systests	`24.87% <23.18%> (?)`
unittests	`72.30% <89.85%> (+40.37%)`	⬆️

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

Impacted Files	Coverage Δ
...ava/org/apache/pulsar/broker/service/Consumer.java	`86.69% <75.00%> (+23.55%)`	⬆️
...va/org/apache/pulsar/broker/service/ServerCnx.java	`71.56% <80.00%> (+32.31%)`	⬆️
...apache/pulsar/broker/web/AuthenticationFilter.java	`92.30% <84.61%> (+27.30%)`	⬆️
...dbalance/extensions/ExtensibleLoadManagerImpl.java	`78.45% <92.30%> (+76.15%)`	⬆️
...xtensions/channel/ServiceUnitStateChannelImpl.java	`84.42% <100.00%> (+83.87%)`	⬆️
...sar/broker/loadbalance/impl/LoadManagerShared.java	`82.10% <100.00%> (+52.10%)`	⬆️
...ache/pulsar/broker/namespace/NamespaceService.java	`71.49% <100.00%> (+28.65%)`	⬆️
...unctions/runtime/kubernetes/KubernetesRuntime.java	`49.16% <100.00%> (+49.16%)`	⬆️

... and 1513 files with indirect coverage changes

lifepuzzlefun · 2023-06-15T13:50:19Z

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

+                        topicName, subscription, status.cause() == null ? "" : status.cause().getMessage(),
+                        totalMessages, this.toString(), status.cause());
+                // If the health check fail, this connection will be closed.
+                cnx.healthCheckManually();


I notice there is a recent added code also check the connection liveness. do you think we can reuse this method ?
see

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

Lines 3395 to 3418 in d85736c

public CompletableFuture<Boolean> checkConnectionLiveness() {

if (connectionLivenessCheckTimeoutMillis > 0) {

return NettyFutureUtil.toCompletableFuture(ctx.executor().submit(() -> {

if (connectionCheckInProgress != null) {

return connectionCheckInProgress;

} else {

final CompletableFuture<Boolean> finalConnectionCheckInProgress = new CompletableFuture<>();

connectionCheckInProgress = finalConnectionCheckInProgress;

ctx.executor().schedule(() -> {

if (finalConnectionCheckInProgress == connectionCheckInProgress

&& !finalConnectionCheckInProgress.isDone()) {

log.warn("[{}] Connection check timed out. Closing connection.", remoteAddress);

ctx.close();

}

}, connectionLivenessCheckTimeoutMillis, TimeUnit.MILLISECONDS);

sendPing();

return finalConnectionCheckInProgress;

}

})).thenCompose(java.util.function.Function.identity());

} else {

// check is disabled

return CompletableFuture.completedFuture((Boolean) null);

}

}

It is a good suggestion. The method checkConnectionLiveness was called when a new consumer registered, it always calls sendPing. Since sending messages to the client is a high-frequency event. So if a connection is inactive, this results in a large number of tasks being pushed into Netty's task pool. We should wrap the method checkConnectionLiveness as a new method to limit the number of tasks. such as

public void healthCheckManually( ) { if (canSendPingNow()) { checkConnectionLiveness(); } }

But the wrap method is not needed to switch thread to the netty thread, it already runs in the netty thread, so calling sendPing directly is better than calling checkConnectionLiveness. The current implementation is the result of final optimization.

lifepuzzlefun · 2023-06-15T13:54:11Z

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

@@ -1783,6 +1789,10 @@ protected void handleAck(CommandAck ack) {
                }
                return null;
            });
+        } else {
+            log.debug("Consumer future is not complete(not complete or error), but received command ack. so discard"


if (log.isDebugEnabled())

lifepuzzlefun

LGTM

codelipenghui

Is it possible to add a test to ensure the connection will be closed immediately if failed to write data to the channel.

codelipenghui · 2023-06-27T10:13:01Z

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

@@ -361,6 +361,12 @@ public Future<Void> sendMessages(final List<? extends Entry> entries, EntryBatch
                msgOutCounter.add(totalMessages);
                bytesOutCounter.add(totalBytes);
                chunkedMessageRate.recordMultipleEvents(totalChunkedMessages, 0);
+            } else {
+                log.warn("[{}-{}] Sent messages to client fail by IO exception[{}], these messages(messages count:"


Is it better to change the log level to debug? It's a message level log, for a cluster with high consumption throughput, it will introduce too many warning logs. We should be careful to add logs on the message level with >= info level.

codelipenghui · 2023-06-27T10:22:43Z

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

+            return;
+        }
+        if (LAST_MANUAL_HEARTBEAT_CHECK_TIME_UPDATER.compareAndSet(this, lastCheckTime, System.currentTimeMillis())) {
+            sendPing();


I'm not sure how it can resolve the problem. If the ping command is sent, but no pong response from the client side, does the broker need to wait for the keep-alive timeout? But the keep alive task already handled it, right?

Or maybe we should call handleKeepAliveTimeout() so that the connection can be closed forcelly without waiting for the response.

Is it better to change the log level to debug? It's a message level log, for a cluster with high consumption throughput, it will introduce too many warning logs. We should be careful to add logs on the message level with >= info level.

Fixed.

Is it possible to add a test to ensure the connection will be closed immediately if failed to write data to the channel?

After I wrote a test, I noticed that the logic of "close connection after the write failure" already exists^[1], I removed the logic which tries to do the same thing. Now the modification of this PR is the following two points:

(important) Since cnx.address + consumerId is the identifier of one consumer; add consumer-id into the log when doing subscribe.

add a test to confirm that even if the error occurs when sending messages to the client, the consumption is still OK.

I have renamed the title of this PR.

Thanks very much. It is an important review.

[1]: https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java#L422

…a to client

…#20568) - Since `cnx.address + consumerId` is the identifier of one consumer; add `consumer-id` into the log when doing subscribe. - add a test to confirm that even if the error occurs when sending messages to the client, the consumption is still OK. - print debug log if ack-command was discarded due to `ConsumerFuture is not complete.` - print debug log if sending a message to the client is failed. (cherry picked from commit a41ac49)

…#20568) - Since `cnx.address + consumerId` is the identifier of one consumer; add `consumer-id` into the log when doing subscribe. - add a test to confirm that even if the error occurs when sending messages to the client, the consumption is still OK. - print debug log if ack-command was discarded due to `ConsumerFuture is not complete.` - print debug log if sending a message to the client is failed.

…apache#20568) - Since `cnx.address + consumerId` is the identifier of one consumer; add `consumer-id` into the log when doing subscribe. - add a test to confirm that even if the error occurs when sending messages to the client, the consumption is still OK. - print debug log if ack-command was discarded due to `ConsumerFuture is not complete.` - print debug log if sending a message to the client is failed. (cherry picked from commit a41ac49) (cherry picked from commit e25d764)

…#20568) - Since `cnx.address + consumerId` is the identifier of one consumer; add `consumer-id` into the log when doing subscribe. - add a test to confirm that even if the error occurs when sending messages to the client, the consumption is still OK. - print debug log if ack-command was discarded due to `ConsumerFuture is not complete.` - print debug log if sending a message to the client is failed.

github-actions bot added the doc-not-needed Your PR changes do not impact docs label Jun 13, 2023

poorbarcode requested review from merlimat, Technoboy- and codelipenghui June 13, 2023 08:46

poorbarcode changed the title ~~[improve][log] Improve dispatcher log to trace io error that may cause consumption stuck~~ [improve][broker] Improve dispatcher log to trace io error that may cause consumption stuck Jun 13, 2023

poorbarcode self-assigned this Jun 13, 2023

poorbarcode added release/2.10.5 release/2.11.2 release/3.0.2 labels Jun 13, 2023

poorbarcode added this to the 3.1.0 milestone Jun 13, 2023

Technoboy- closed this Jun 13, 2023

Technoboy- reopened this Jun 13, 2023

poorbarcode changed the title ~~[improve][broker] Improve dispatcher log to trace io error that may cause consumption stuck~~ [improve] [broker] Improve dispatcher log to trace io error that may cause consumption stuck Jun 13, 2023

poorbarcode added the ready-to-test label Jun 13, 2023

Technoboy- approved these changes Jun 13, 2023

View reviewed changes

lifepuzzlefun reviewed Jun 13, 2023

View reviewed changes

poorbarcode requested a review from lifepuzzlefun June 13, 2023 16:05

poorbarcode mentioned this pull request Jun 13, 2023

[fix] [broker]Consumption stuck due to orphan consumers after the connection reconnect #20573

Closed

4 tasks

mattisonchao reviewed Jun 15, 2023

View reviewed changes

poorbarcode requested a review from mattisonchao June 15, 2023 04:37

poorbarcode changed the title ~~[improve] [broker] Improve dispatcher log to trace io error that may cause consumption stuck~~ [improve] [broker] Close connection if got an IO error when sending messages to the client Jun 15, 2023

lifepuzzlefun reviewed Jun 15, 2023

View reviewed changes

poorbarcode requested a review from lifepuzzlefun June 15, 2023 16:12

lifepuzzlefun approved these changes Jun 17, 2023

View reviewed changes

codelipenghui reviewed Jun 27, 2023

View reviewed changes

poorbarcode changed the title ~~[improve] [broker] Close connection if got an IO error when sending messages to the client~~ [improve] [broker] Add consumer-id into the log when doing subscribe. Jun 27, 2023

poorbarcode requested a review from codelipenghui June 27, 2023 11:34

congbobo184 approved these changes Jul 10, 2023

View reviewed changes

poorbarcode added 12 commits July 10, 2023 14:21

[improve][log] Improve dispatcher log to trace io error when send dat…

41d50cf

…a to client

-

bde1358

-

6e6487a

address comments

0b5174a

improve log

8ef79c4

add warn log when discard ack command

3e8cabc

change log level of discard ack to debug

a94c201

do health check manually when sent messages to client fail

3b3ce94

remove unnecessary imports

e4d0335

add a check if (log.isDebugEnabled()) {

6e45b23

address comments

19ebf6b

remove unnecessary imports

1937acd

poorbarcode force-pushed the improve/consumer_log branch from beb8650 to 1937acd Compare July 10, 2023 06:21

poorbarcode merged commit a41ac49 into apache:master Jul 12, 2023
45 checks passed

Technoboy- added release/2.11.3 cherry-picked/branch-2.11 and removed release/2.11.2 labels Jul 17, 2023

Technoboy- added the cherry-picked/branch-3.0 label Aug 17, 2023

Technoboy- added release/2.10.6 and removed release/2.10.5 labels Dec 4, 2023

lhotari added the cherry-picked/branch-2.10 label Mar 5, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[improve] [broker] Add consumer-id into the log when doing subscribe. #20568

[improve] [broker] Add consumer-id into the log when doing subscribe. #20568

poorbarcode commented Jun 13, 2023 •

edited

Loading

lifepuzzlefun Jun 13, 2023

poorbarcode Jun 13, 2023

codelipenghui Jun 27, 2023

mattisonchao Jun 15, 2023 •

edited

Loading

poorbarcode Jun 15, 2023

mattisonchao Jun 15, 2023

poorbarcode Jun 15, 2023

codecov-commenter commented Jun 15, 2023

lifepuzzlefun Jun 15, 2023

poorbarcode Jun 15, 2023 •

edited

Loading

lifepuzzlefun Jun 15, 2023

poorbarcode Jun 15, 2023

lifepuzzlefun left a comment

codelipenghui left a comment

codelipenghui Jun 27, 2023

codelipenghui Jun 27, 2023

poorbarcode Jun 27, 2023

	public CompletableFuture<Boolean> checkConnectionLiveness() {
	if (connectionLivenessCheckTimeoutMillis > 0) {
	return NettyFutureUtil.toCompletableFuture(ctx.executor().submit(() -> {
	if (connectionCheckInProgress != null) {
	return connectionCheckInProgress;
	} else {
	final CompletableFuture<Boolean> finalConnectionCheckInProgress = new CompletableFuture<>();
	connectionCheckInProgress = finalConnectionCheckInProgress;
	ctx.executor().schedule(() -> {
	if (finalConnectionCheckInProgress == connectionCheckInProgress
	&& !finalConnectionCheckInProgress.isDone()) {
	log.warn("[{}] Connection check timed out. Closing connection.", remoteAddress);
	ctx.close();
	}
	}, connectionLivenessCheckTimeoutMillis, TimeUnit.MILLISECONDS);
	sendPing();
	return finalConnectionCheckInProgress;
	}
	})).thenCompose(java.util.function.Function.identity());
	} else {
	// check is disabled
	return CompletableFuture.completedFuture((Boolean) null);
	}
	}

[improve] [broker] Add consumer-id into the log when doing subscribe. #20568

[improve] [broker] Add consumer-id into the log when doing subscribe. #20568

Conversation

poorbarcode commented Jun 13, 2023 • edited Loading

Modifications

Documentation

Matching PR in forked repository

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mattisonchao Jun 15, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov-commenter commented Jun 15, 2023

Codecov Report

Choose a reason for hiding this comment

poorbarcode Jun 15, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lifepuzzlefun left a comment

Choose a reason for hiding this comment

codelipenghui left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

poorbarcode commented Jun 13, 2023 •

edited

Loading

mattisonchao Jun 15, 2023 •

edited

Loading

poorbarcode Jun 15, 2023 •

edited

Loading