Skip to content
This repository was archived by the owner on May 8, 2026. It is now read-only.

Commit 0484400

Browse files
SamuelSchaubSamuelSchaub
committed
fix: create Logger objects for each sub-system tag
1 parent 8274edb commit 0484400

4 files changed

Lines changed: 110 additions & 29 deletions

File tree

‎google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/LoggingUtil.java‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
1+
/*
2+
* Copyright 2016 Google LLC
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package com.google.cloud.pubsub.v1;
18+
19+
import com.google.pubsub.v1.PubsubMessage;
20+
21+
final class LoggingUtil {
22+
private LoggingUtil() {
23+
}
24+
25+
static String getLogPrefix(PubsubMessageWrapper messageWrapper, String ackId, boolean exactlyOnceDeliveryEnabled) {
26+
if (messageWrapper == null || messageWrapper.getPubsubMessage() == null) {
27+
return " Ack ID: " + ackId + ", Exactly Once Delivery: " + exactlyOnceDeliveryEnabled
28+
+ " (Message details not available)";
29+
}
30+
31+
PubsubMessage message = messageWrapper.getPubsubMessage();
32+
String messageId = message.getMessageId();
33+
String orderingKey = message.getOrderingKey();
34+
35+
StringBuilder sb = new StringBuilder();
36+
sb.append("Message ID: ").append(messageId);
37+
sb.append(", Ack ID: ").append(ackId);
38+
if (orderingKey != null && !orderingKey.isEmpty()) {
39+
sb.append(", Ordering Key: ").append(orderingKey);
40+
}
41+
sb.append(", Exactly Once Delivery: ").append(exactlyOnceDeliveryEnabled);
42+
return sb.toString();
43+
}
44+
45+
static String getPublisherLogPrefix(PubsubMessageWrapper messageWrapper) {
46+
if (messageWrapper == null || messageWrapper.getPubsubMessage() == null) {
47+
return " (Message details not available)";
48+
}
49+
50+
PubsubMessage message = messageWrapper.getPubsubMessage();
51+
String messageId = message.getMessageId();
52+
String orderingKey = message.getOrderingKey();
53+
54+
StringBuilder sb = new StringBuilder();
55+
sb.append("Message ID: ").append(messageId);
56+
if (orderingKey != null && !orderingKey.isEmpty()) {
57+
sb.append(", Ordering Key: ").append(orderingKey);
58+
}
59+
return sb.toString();
60+
}
61+
}

‎google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java‎

Lines changed: 37 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,13 @@
5959
*/
6060
class MessageDispatcher {
6161
private static final Logger logger = Logger.getLogger(MessageDispatcher.class.getName());
62+
private static final Logger slowAckLogger = Logger.getLogger("slow-ack");
63+
private static final Logger callbackDeliveryLogger = Logger.getLogger("callback-delivery");
64+
private static final Logger expiryLogger = Logger.getLogger("expiry");
65+
private static final Logger callbackExceptionsLogger = Logger.getLogger("callback-exceptions");
66+
private static final Logger ackBatchLogger = Logger.getLogger("ack-batch");
67+
private static final Logger subscriberFlowControlLogger = Logger.getLogger("subscriber-flow-control");
68+
private static final Logger ackNackLogger = Logger.getLogger("ack-nack");
6269

6370
@InternalApi static final double PERCENTILE_FOR_ACK_DEADLINE_UPDATES = 99.9;
6471
@InternalApi static final Duration PENDING_ACKS_SEND_DELAY = Duration.ofMillis(100);
@@ -159,14 +166,11 @@ private void forget() {
159166

160167
@Override
161168
public void onFailure(Throwable t) {
162-
logger.log(
163-
Level.WARNING,
164-
"pubsub:callback-exception - MessageReceiver failed to process ack ID: "
165-
+ this.ackRequestData.getAckId()
166-
+ ", the message will be nacked."
167-
+ " Message ID: "
168-
+ this.ackRequestData.getMessageWrapper().getPubsubMessage().getMessageId(),
169-
t);
169+
if (callbackExceptionsLogger.isLoggable(Level.WARNING)) {
170+
String prefix = LoggingUtil.getLogPrefix(this.ackRequestData.getMessageWrapper(),
171+
this.getAckRequestData().getAckId(), exactlyOnceDeliveryEnabled.get());
172+
callbackExceptionsLogger.log(Level.WARNING, "pubsub:callback-exceptions - MessageReceiver exception. " + prefix, t);
173+
}
170174
this.ackRequestData.setResponse(AckResponse.OTHER, false);
171175
pendingNacks.add(this.ackRequestData);
172176
tracer.endSubscribeProcessSpan(this.ackRequestData.getMessageWrapper(), "nack");
@@ -177,9 +181,13 @@ public void onFailure(Throwable t) {
177181
public void onSuccess(AckReply reply) {
178182
int ackLatency =
179183
Ints.saturatedCast((long) Math.ceil((clock.millisTime() - receivedTimeMillis) / 1000D));
180-
String messageId = this.ackRequestData.getMessageWrapper().getPubsubMessage().getMessageId();
184+
String logPrefix = "";
185+
if (slowAckLogger.isLoggable(Level.FINE) || ackNackLogger.isLoggable(Level.FINE)) {
186+
logPrefix = LoggingUtil.getLogPrefix(this.ackRequestData.getMessageWrapper(),
187+
this.ackRequestData.getAckId(), exactlyOnceDeliveryEnabled.get());
188+
}
181189
if (ackLatency >= ackLatencyDistribution.getPercentile(slowAckPercentile)) {
182-
logger.log(Level.FINE, "pubsub:slow-ack - Message ID: {0}", messageId);
190+
slowAckLogger.log(Level.FINE, "pubsub:slow-ack - " + logPrefix);
183191
}
184192

185193
switch (reply) {
@@ -194,12 +202,12 @@ public void onSuccess(AckReply reply) {
194202
ackLatencyDistribution.record(ackLatency);
195203
tracer.endSubscribeProcessSpan(this.ackRequestData.getMessageWrapper(), "ack");
196204
}
197-
logger.log(Level.FINE, "pubsub:ack-nack - Action: ACK - Message ID: {0}", messageId);
205+
ackNackLogger.log(Level.FINE, "pubsub:ack-nack - " + logPrefix + " - Action: ACK");
198206
break;
199207
case NACK:
200208
pendingNacks.add(this.ackRequestData);
201209
tracer.endSubscribeProcessSpan(this.ackRequestData.getMessageWrapper(), "nack");
202-
logger.log(Level.FINE, "pubsub:ack-nack - Action: NACK - Message ID: {0}", messageId);
210+
ackNackLogger.log(Level.FINE, "pubsub:ack-nack - " + logPrefix + " - Action: NACK");
203211
break;
204212
default:
205213
throw new IllegalArgumentException(String.format("AckReply: %s not supported", reply));
@@ -577,17 +585,20 @@ private void processBatch(List<OutstandingMessage> batch) {
577585
for (OutstandingMessage message : batch) {
578586
// This is a blocking flow controller. We have already incremented messagesWaiter, so
579587
// shutdown will block on processing of all these messages anyway.
588+
String logPrefix = "";
589+
if (subscriberFlowControlLogger.isLoggable(Level.FINE)) {
590+
logPrefix = LoggingUtil.getLogPrefix(message.messageWrapper(),
591+
message.messageWrapper().getAckId(), exactlyOnceDeliveryEnabled.get());
592+
}
580593
tracer.startSubscribeConcurrencyControlSpan(message.messageWrapper());
581594
try {
582-
logger.log(Level.FINE, "pubsub:subscriber-flow-control - Flow controller is blocking.");
595+
subscriberFlowControlLogger.log(Level.FINE, "pubsub:subscriber-flow-control - " + logPrefix + " - Flow controller is blocking.");
583596
flowController.reserve(1, message.messageWrapper().getPubsubMessage().getSerializedSize());
584-
logger.log(
585-
Level.FINE, "pubsub:subscriber-flow-control - Flow controller is done blocking.");
597+
subscriberFlowControlLogger.log(Level.FINE, "pubsub:subscriber-flow-control - " + logPrefix + " - Flow controller is done blocking.");
586598
tracer.endSubscribeConcurrencyControlSpan(message.messageWrapper());
587599
} catch (FlowControlException unexpectedException) {
588600
// This should be a blocking flow controller and never throw an exception.
589-
logger.log(
590-
Level.FINE, "pubsub:subscriber-flow-control - Flow controller unexpected exception.");
601+
subscriberFlowControlLogger.log(Level.FINE, "pubsub:subscriber-flow-control - " + logPrefix + " - Flow controller unexpected exception.");
591602
tracer.setSubscribeConcurrencyControlSpanException(
592603
message.messageWrapper(), unexpectedException);
593604
throw new IllegalStateException("Flow control unexpected exception", unexpectedException);
@@ -626,6 +637,11 @@ private void processOutstandingMessage(final AckHandler ackHandler) {
626637
@Override
627638
public void run() {
628639
try {
640+
String logPrefix = "";
641+
if (expiryLogger.isLoggable(Level.FINE) || callbackDeliveryLogger.isLoggable(Level.FINE)) {
642+
logPrefix = LoggingUtil.getLogPrefix(messageWrapper,
643+
ackHandler.ackRequestData.getAckId(), exactlyOnceDeliveryEnabled.get());
644+
}
629645
if (ackHandler
630646
.totalExpiration
631647
.plusSeconds(messageDeadlineSeconds.get())
@@ -635,12 +651,11 @@ public void run() {
635651
// Don't nack it either, because we'd be nacking someone else's message.
636652
ackHandler.forget();
637653
tracer.setSubscriberSpanExpirationResult(messageWrapper);
638-
logger.log(Level.FINE, "pubsub:expiry - Message ID: {0}", message.getMessageId());
654+
expiryLogger.log(Level.FINE, "pubsub:expiry - " + logPrefix);
639655
return;
640656
}
641657
tracer.startSubscribeProcessSpan(messageWrapper);
642-
logger.log(
643-
Level.FINE, "pubsub:callback-delivery - Message ID: {0}", message.getMessageId());
658+
callbackDeliveryLogger.log(Level.FINE, "pubsub:callback-delivery - " + logPrefix);
644659
if (shouldSetMessageFuture()) {
645660
// This is the message future that is propagated to the user
646661
SettableApiFuture<AckResponse> messageFuture =
@@ -744,7 +759,6 @@ void processOutstandingOperations() {
744759
if (!nackRequestDataList.isEmpty()) {
745760
modackRequestData.add(new ModackRequestData(0, nackRequestDataList));
746761
}
747-
logger.log(Level.FINER, "pubsub:ack-batch - Sending {0} nacks", nackRequestDataList.size());
748762

749763
List<AckRequestData> ackRequestDataReceipts = new ArrayList<AckRequestData>();
750764
pendingReceipts.drainTo(ackRequestDataReceipts);
@@ -754,13 +768,13 @@ void processOutstandingOperations() {
754768
receiptModack.setIsReceiptModack(true);
755769
modackRequestData.add(receiptModack);
756770
}
757-
logger.log(Level.FINER, "Sending {0} receipts", ackRequestDataReceipts.size());
758771

759772
ackProcessor.sendModackOperations(modackRequestData);
760773

761774
List<AckRequestData> ackRequestDataList = new ArrayList<AckRequestData>();
762775
pendingAcks.drainTo(ackRequestDataList);
763-
logger.log(Level.FINER, "pubsub:ack-bbatch - Sending {0} acks", ackRequestDataList.size());
776+
ackBatchLogger.log(Level.FINE, "pubsub:ack-batch - Sending {0} ACKs, {1} NACKs, {2} receipts. Exactly Once Delivery: {3}",
777+
new Object[]{ackRequestDataList.size(), nackRequestDataList.size(), ackRequestDataList.size(), exactlyOnceDeliveryEnabled.get()});
764778

765779
ackProcessor.sendAckOperations(ackRequestDataList);
766780
}

‎google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Publisher.java‎

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -94,6 +94,7 @@
9494
*/
9595
public class Publisher implements PublisherInterface {
9696
private static final Logger logger = Logger.getLogger(Publisher.class.getName());
97+
private static final Logger publishBatchLogger = Logger.getLogger("publish-batch");
9798

9899
private static final String GZIP_COMPRESSION = "gzip";
99100

@@ -509,10 +510,14 @@ private void publishOutstandingBatch(final OutstandingBatch outstandingBatch) {
509510
logger.log(Level.WARNING, "Attempted to publish batch with zero messages.");
510511
return;
511512
}
512-
logger.log(
513-
Level.FINE,
514-
"pubsub:publish-batch - Attempting to publish batch {0} messages.",
515-
outstandingBatch.size());
513+
514+
if (publishBatchLogger.isLoggable(Level.FINE)) {
515+
String logPrefix = LoggingUtil.getPublisherLogPrefix(outstandingBatch.getMessageWrappers().get(0));
516+
publishBatchLogger.log(
517+
Level.FINE,
518+
"pubsub:publish-batch - " + logPrefix + " - Attempting to publish batch {0} messages.",
519+
outstandingBatch.size());
520+
}
516521

517522
final ApiFutureCallback<PublishResponse> futureCallback =
518523
new ApiFutureCallback<PublishResponse>() {

‎google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -77,6 +77,7 @@
7777
final class StreamingSubscriberConnection extends AbstractApiService implements AckProcessor {
7878
private static final Logger logger =
7979
Logger.getLogger(StreamingSubscriberConnection.class.getName());
80+
private static final Logger subscriberStreamsLogger = Logger.getLogger("subscriber-streams");
8081

8182
private static final Duration INITIAL_CHANNEL_RECONNECT_BACKOFF = Duration.ofMillis(100);
8283
private static final Duration MAX_CHANNEL_RECONNECT_BACKOFF = Duration.ofSeconds(10);
@@ -222,15 +223,15 @@ public boolean getExactlyOnceDeliveryEnabled() {
222223
@Override
223224
protected void doStart() {
224225
logger.config("Starting subscriber.");
225-
logger.log(Level.FINER, "pubsub:subscriber-streams - Opening stream.");
226+
subscriberStreamsLogger.log(Level.FINE, "pubsub:subscriber-streams - Opening stream.");
226227
messageDispatcher.start();
227228
initialize();
228229
notifyStarted();
229230
}
230231

231232
@Override
232233
protected void doStop() {
233-
logger.log(Level.FINER, "pubsub:subscriber-streams - Closing stream.");
234+
subscriberStreamsLogger.log(Level.FINE, "pubsub:subscriber-streams - Closing stream.");
234235
lock.lock();
235236
try {
236237
clientStream.closeSendWithError(Status.CANCELLED.asException());

0 commit comments

Comments
 (0)