refactor: Add debug logging (#741)

Author: tmdiep

google-cloud-pubsublite/src/main/java/com/google/cloud/pubsublite/internal/wire/AssignerImpl.java

@@ -28,12 +28,15 @@
 import com.google.cloud.pubsublite.proto.PartitionAssignmentRequest;
 import com.google.cloud.pubsublite.v1.PartitionAssignmentServiceClient;
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.flogger.GoogleLogger;
 import com.google.errorprone.annotations.concurrent.GuardedBy;
 import java.util.HashSet;
 import java.util.Set;
 
 public class AssignerImpl extends TrivialProxyService
  implements Assigner, RetryingConnectionObserver<PartitionAssignment> {
+ private static final GoogleLogger logger = GoogleLogger.forEnclosingClass();
+
  private final PartitionAssignmentRequest initialRequest;
 
  private final CloseableMonitor monitor = new CloseableMonitor();
@@ -90,7 +93,9 @@ private static Set<Partition> toSet(PartitionAssignment assignment) throws ApiEx
  @Override
  public void onClientResponse(PartitionAssignment value) throws CheckedApiException {
  try (CloseableMonitor.Hold h = monitor.enter()) {
- receiver.handleAssignment(toSet(value));
+ Set<Partition> partitions = toSet(value);
+ receiver.handleAssignment(partitions);
+ logger.atInfo().log("Subscribed to partitions: %s", partitions);
  connection.modifyConnection(connectionOr -> connectionOr.ifPresent(ConnectedAssigner::ack));
  }
  }

google-cloud-pubsublite/src/main/java/com/google/cloud/pubsublite/internal/wire/PublisherImpl.java

@@ -41,6 +41,7 @@
 import com.google.cloud.pubsublite.v1.PublisherServiceClient;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
+import com.google.common.flogger.GoogleLogger;
 import com.google.common.util.concurrent.Monitor;
 import java.util.ArrayDeque;
 import java.util.Collection;
@@ -54,6 +55,8 @@
 
 public final class PublisherImpl extends ProxyService
  implements Publisher<Offset>, RetryingConnectionObserver<Offset> {
+ private static final GoogleLogger logger = GoogleLogger.forEnclosingClass();
+
  private final BatchingSettings batchingSettings;
  private final PublishRequest initialRequest;
  private Future<?> alarmFuture;
@@ -135,6 +138,7 @@ private void rebatchForRestart() {
  messages.add(UnbatchedMessage.of(batch.messages.get(i), batch.messageFutures.get(i)));
  }
  }
+ logger.atFiner().log("Re-publishing %s messages after reconnection", messages.size());
  long size = 0;
  int count = 0;
  Queue<UnbatchedMessage> currentBatch = new ArrayDeque<>();

google-cloud-pubsublite/src/main/java/com/google/cloud/pubsublite/internal/wire/RetryingConnectionImpl.java

@@ -102,7 +102,9 @@ public void reinitialize(StreamRequestT initialRequest) {
  try (CloseableMonitor.Hold h = connectionMonitor.enter()) {
  if (completed) return;
  lastInitialRequest = initialRequest;
+ logger.atFiner().log("Start initializing connection for %s", streamDescription());
  currentConnection = connectionFactory.New(streamFactory, this, lastInitialRequest);
+ logger.atFiner().log("Initialized connection for %s", streamDescription());
  }
  }
 
@@ -111,18 +113,15 @@ protected void doStop() {
  try (CloseableMonitor.Hold h = connectionMonitor.enter()) {
  if (completed) return;
  completed = true;
- logger.atFine().log(
- String.format("Terminating connection with initial request %s.", streamDescription()));
+ logger.atFine().log("Terminating connection for %s", streamDescription());
  currentConnection.close();
  } catch (Throwable t) {
  logger.atWarning().withCause(t).log(
- String.format(
- "Failed while terminating connection with initial request %s.", streamDescription()));
+ "Failed while terminating connection for %s", streamDescription());
  notifyFailed(t);
  return;
  }
- logger.atFine().log(
- String.format("Terminated connection with initial request %s.", streamDescription()));
+ logger.atFine().log("Terminated connection for %s", streamDescription());
  notifyStopped();
  }
 
@@ -196,7 +195,8 @@ public final void onError(Throwable t) {
  return;
  }
  logger.atFine().withCause(t).log(
- "Stream disconnected attempting retry, after %s milliseconds", backoffTime);
+ "Stream disconnected attempting retry, after %s milliseconds for %s",
+ backoffTime, streamDescription());
  ScheduledFuture<?> retry =
  SystemExecutors.getAlarmExecutor()
  .schedule(
@@ -214,7 +214,7 @@ public final void onError(Throwable t) {
 
  @Override
  public final void onComplete() {
- logger.atFine().log("Stream completed for %s.", streamDescription());
+ logger.atFine().log("Stream completed for %s", streamDescription());
  boolean expectedCompletion;
  try (CloseableMonitor.Hold h = connectionMonitor.enter()) {
  expectedCompletion = completed;
@@ -227,7 +227,7 @@ public final void onComplete() {
 
  private String streamDescription() {
  try (CloseableMonitor.Hold h = connectionMonitor.enter()) {
- return lastInitialRequest.toString();
+ return lastInitialRequest.getClass().getSimpleName() + ": " + lastInitialRequest.toString();
  }
  }
 }