fix: retry Cancelled stream errors (#1124)

tmdiep · web-flow · commit 6d54da81d1bf · 2022-04-26T09:24:38.000-04:00
* fix: retry Cancelled stream errors

* Fix lint
diff --git a/google-cloud-pubsublite/src/main/java/com/google/cloud/pubsublite/ErrorCodes.java b/google-cloud-pubsublite/src/main/java/com/google/cloud/pubsublite/ErrorCodes.java
@@ -28,7 +28,8 @@ public final class ErrorCodes {
           Code.INTERNAL,
           Code.UNAVAILABLE,
           Code.UNKNOWN,
-          Code.RESOURCE_EXHAUSTED);
+          Code.RESOURCE_EXHAUSTED,
+          Code.CANCELLED);
 
   public static boolean IsRetryableForStreams(Code code) {
     return STREAM_RETRYABLE_CODES.contains(code);
diff --git a/google-cloud-pubsublite/src/main/java/com/google/cloud/pubsublite/internal/wire/PublisherImpl.java b/google-cloud-pubsublite/src/main/java/com/google/cloud/pubsublite/internal/wire/PublisherImpl.java
@@ -145,7 +145,9 @@ 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());
+    logger.atFiner().log(
+        "Re-publishing %s messages after reconnection for partition %s",
+        messages.size(), initialRequest.getInitialRequest().getPartition());
     long size = 0;
     int count = 0;
     Queue<UnbatchedMessage> currentBatch = new ArrayDeque<>();
diff --git a/google-cloud-pubsublite/src/main/java/com/google/cloud/pubsublite/internal/wire/RetryingConnectionImpl.java b/google-cloud-pubsublite/src/main/java/com/google/cloud/pubsublite/internal/wire/RetryingConnectionImpl.java
@@ -104,7 +104,7 @@ public void reinitialize(StreamRequestT initialRequest) {
       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());
+      logger.atFiner().log("Finished initializing connection for %s", streamDescription());
     }
   }
 
@@ -123,7 +123,6 @@ protected void doStop() {
       notifyFailed(t);
       return;
     }
-    logger.atFine().log("Terminated connection for %s", streamDescription());
     notifyStopped();
   }
 
@@ -144,6 +143,7 @@ void setPermanentError(Throwable error) {
       if (completed) return;
       completed = true;
     }
+    logger.atInfo().withCause(error).log("Permanent error occurred for %s", streamDescription());
     notifyFailed(error);
   }
 

Original file line number	Diff line number	Diff line change
`@@ -145,7 +145,9 @@ private void rebatchForRestart() {`
`145`	`145`	`messages.add(UnbatchedMessage.of(batch.messages.get(i), batch.messageFutures.get(i)));`
`146`	`146`	`}`
`147`	`147`	`}`
`148`		`- logger.atFiner().log("Re-publishing %s messages after reconnection", messages.size());`
	`148`	`+ logger.atFiner().log(`
	`149`	`+ "Re-publishing %s messages after reconnection for partition %s",`
	`150`	`+ messages.size(), initialRequest.getInitialRequest().getPartition());`
`149`	`151`	`long size = 0;`
`150`	`152`	`int count = 0;`
`151`	`153`	`Queue<UnbatchedMessage> currentBatch = new ArrayDeque<>();`
Original file line number	Diff line number	Diff line change
`@@ -104,7 +104,7 @@ public void reinitialize(StreamRequestT initialRequest) {`
`104`	`104`	`lastInitialRequest = initialRequest;`
`105`	`105`	`logger.atFiner().log("Start initializing connection for %s", streamDescription());`
`106`	`106`	`currentConnection = connectionFactory.New(streamFactory, this, lastInitialRequest);`
`107`		`- logger.atFiner().log("Initialized connection for %s", streamDescription());`
	`107`	`+ logger.atFiner().log("Finished initializing connection for %s", streamDescription());`
`108`	`108`	`}`
`109`	`109`	`}`
`110`	`110`
`@@ -123,7 +123,6 @@ protected void doStop() {`
`123`	`123`	`notifyFailed(t);`
`124`	`124`	`return;`
`125`	`125`	`}`
`126`		`- logger.atFine().log("Terminated connection for %s", streamDescription());`
`127`	`126`	`notifyStopped();`
`128`	`127`	`}`
`129`	`128`
`@@ -144,6 +143,7 @@ void setPermanentError(Throwable error) {`
`144`	`143`	`if (completed) return;`
`145`	`144`	`completed = true;`
`146`	`145`	`}`
	`146`	`+ logger.atInfo().withCause(error).log("Permanent error occurred for %s", streamDescription());`
`147`	`147`	`notifyFailed(error);`
`148`	`148`	`}`
`149`	`149`