René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"Pubsub: Unbounded retry forced when publishing with message ordering enabled","articleBody":"#### Summary\r\n\r\nWhen publishing to pubsub with message ordering enabled, the library overrides the gRPC retry settings and forces unbounded retry. The application using this library is then unable to detect failures. The workaround is to add timeout when waiting for the returned future, but this makes the retry/timeout policy scattered and surprising to configure.\r\n\r\nThis is the code that enforces infinite retry.\r\nhttps://github.com/googleapis/java-pubsub/blob/main/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Publisher.java#L168 line 174 to line 180\r\n\r\nThe exact reason why the infinite retry does not eventually become successful is unclear. We have multiple random occurrences where applications running in a GKE cluster would stop publishing. In some instances, we had even waited for the application to retry for up to a week without success. Unfortunately we do not have a reliable way to trigger this type of failure and we do not have the low level gRPC logs when the failure happened.\r\n\r\nMy suggestion is to allow the library user to set a sane upper bound for retry, and let failures be propagated through the `onFailure` callback.\r\n\r\n#### Environment details\r\n\r\n1. API: Pubsub\r\n2. OS type and version: Ubuntu 22.04\r\n3. Java version: temurin-17.0.8\r\n4.  version(s): google-cloud-pubsub-1.123.12\r\n\r\n#### Steps to reproduce\r\n\r\n  1. Enable message ordering in publisher\r\n  2. Publish messages\r\n  3. Simulate failure, such as network disconnect\r\n\r\n#### Code example\r\n\r\nBased on sample publish code in documentation.\r\n\r\n```java\r\n    Publisher publisher =\r\n        Publisher.newBuilder(topicName)\r\n            // Sending messages to the same region ensures they are received in order\r\n            // even when multiple publishers are used.\r\n            .setEndpoint(\"us-east1-pubsub.googleapis.com:443\")\r\n            .setEnableMessageOrdering(true)\r\n            .build();\r\n\r\n    try {\r\n      Map\u003cString, String\u003e messages = new LinkedHashMap\u003cString, String\u003e();\r\n      messages.put(\"message1\", \"key1\");\r\n      // auto-generate many more messages here ...\r\n\r\n      for (Map.Entry\u003cString, String\u003e entry : messages.entrySet()) {\r\n        ByteString data = ByteString.copyFromUtf8(entry.getKey());\r\n        PubsubMessage pubsubMessage =\r\n            PubsubMessage.newBuilder().setData(data).setOrderingKey(entry.getValue()).build();\r\n        ApiFuture\u003cString\u003e future = publisher.publish(pubsubMessage);\r\n\r\n        // Add an asynchronous callback to handle publish success / failure.\r\n        ApiFutures.addCallback(\r\n            future,\r\n            new ApiFutureCallback\u003cString\u003e() {\r\n\r\n              // onFailure is never called, because the publisher will retry indefinitely\r\n\r\n              @Override\r\n              public void onFailure(Throwable throwable) {\r\n                if (throwable instanceof ApiException) {\r\n                  ApiException apiException = ((ApiException) throwable);\r\n                  // Details on the API exception.\r\n                  System.out.println(apiException.getStatusCode().getCode());\r\n                  System.out.println(apiException.isRetryable());\r\n                }\r\n                System.out.println(\"Error publishing message : \" + pubsubMessage.getData());\r\n              }\r\n\r\n              @Override\r\n              public void onSuccess(String messageId) {\r\n                // Once published, returns server-assigned message ids (unique within the topic).\r\n                System.out.println(pubsubMessage.getData() + \" : \" + messageId);\r\n              }\r\n            },\r\n            MoreExecutors.directExecutor());\r\n      }\r\n    } finally {\r\n      // When finished with the publisher, shutdown to free up resources.\r\n      publisher.shutdown();\r\n      publisher.awaitTermination(1, TimeUnit.MINUTES);\r\n    }\r\n```\r\n\r\n#### Stack trace\r\n```\r\nNo stack trace because of infinite retry\r\n```\r\n\r\n#### External references such as API reference guides\r\n\r\n- https://cloud.google.com/pubsub/docs/publisher\r\n\r\n#### Any additional information below\r\n\r\n","author":{"url":"https://github.com/poernahi","@type":"Person","name":"poernahi"},"datePublished":"2023-08-17T17:09:44.000Z","interactionStatistic":{"@type":"InteractionCounter","interactionType":"https://schema.org/CommentAction","userInteractionCount":5},"url":"https://github.com/1702/java-pubsub/issues/1702"}