René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"[pubsub] Subscriber.stopAsync().awaitTerminated() doesn't wait for all acks to be sent","articleBody":"In working on a gRPC emulation service for Cloud Pub/Sub, I was trying to validate correct behavior by publishing a series of messages, receiving and acknowleding the exact same set of messages, and then ensuring that no backlog existed for the subscription.\r\n\r\nHere's the general test pattern:\r\n```\r\nString messagePrefix = subscriptionProperties.getName() + System.currentTimeMillis() + \"-\";\r\nint messages = 5000;\r\nSet\u003cString\u003e messagesSet = new TreeSet\u003c\u003e();\r\nfor (int i = 0; i \u003c messages; i++) {\r\n  messagesSet.add(messagePrefix + i);\r\n}\r\nCountDownLatch countDownLatch = new CountDownLatch(messages);\r\nSet\u003cString\u003e publishedIds = new ConcurrentSkipListSet\u003c\u003e();\r\nMap\u003cString, Integer\u003e receivedIds = new ConcurrentHashMap\u003c\u003e();\r\nfor (String data : messagesSet) {\r\n  PubsubMessage message =\r\n      PubsubMessage.newBuilder().setData(ByteString.copyFromUtf8(data)).build();\r\n  ApiFutures.addCallback(\r\n      publisher.publish(message),\r\n      new ApiFutureCallback\u003cString\u003e() {\r\n        @Override\r\n        public void onFailure(Throwable throwable) {\r\n          LOGGER.warning(\"Error on Publish \" + throwable.getMessage());\r\n        }\r\n\r\n        @Override\r\n        public void onSuccess(String s) {\r\n          publishedIds.add(s);\r\n        }\r\n      });\r\n}\r\npublisher.shutdown();\r\nassertEquals(messages, publishedIds.size());\r\n\r\nsubscriber =\r\n    Subscriber.newBuilder(\r\n            ProjectSubscriptionName.of(PROJECT, subscriptionProperties.getName()),\r\n            (message, consumer) -\u003e {\r\n              consumer.ack();\r\n              LOGGER.fine(\"Received and Acknowledging \" + message.getMessageId());\r\n              if (!receivedIds.containsKey(message.getMessageId())) {\r\n                receivedIds.put(message.getMessageId(), 1);\r\n              } else {\r\n                int current = receivedIds.get(message.getMessageId());\r\n                receivedIds.put(message.getMessageId(), Integer.valueOf(++current));\r\n              }\r\n              countDownLatch.countDown();\r\n            })\r\n        //.setChannelProvider(channelProvider)\r\n        //.setCredentialsProvider(credentialsProvider)\r\n        .build();\r\n\r\npublisher.shutdown();\r\nassertEquals(messages, publishedIds.size());\r\n\r\nsubscriber.startAsync();\r\nLOGGER.info(\"Publisher complete, waiting for 5,000 messages to be received by Subscriber\");\r\ncountDownLatch.await();\r\nLOGGER.info(\"Shutting down Subscriber\");\r\nsubscriber.stopAsync().awaitTerminated();\r\n```\r\n\r\nI executed this test after creating a fresh topic and subscription. My expectation was that after the CountDownLatch reached 0, the Subscriber would shut down and block until all acknowledgements were transmitted.\r\n\r\nIn the client library debug logs, I see the following ack lines, which add up to 5,000 as expected. Interestingly, the final 1,517 acks coming from a thread named *time-limited test*, which is the main thread of my integration test case.\r\n```\r\n2018-03-20 14:10:41.694 [Thread-50] DEBUG com.google.cloud.pubsub.v1.MessageDispatcher : Sending 78 acks\r\n2018-03-20 14:10:41.693 [Thread-42] DEBUG com.google.cloud.pubsub.v1.MessageDispatcher : Sending 70 acks\r\n2018-03-20 14:10:41.696 [Thread-46] DEBUG com.google.cloud.pubsub.v1.MessageDispatcher : Sending 29 acks\r\n2018-03-20 14:10:41.700 [Thread-48] DEBUG com.google.cloud.pubsub.v1.MessageDispatcher : Sending 100 acks\r\n2018-03-20 14:10:41.695 [Thread-52] DEBUG com.google.cloud.pubsub.v1.MessageDispatcher : Sending 169 acks\r\n2018-03-20 14:10:41.702 [Thread-44] DEBUG com.google.cloud.pubsub.v1.MessageDispatcher : Sending 42 acks\r\n2018-03-20 14:10:41.723 [Thread-48] DEBUG com.google.cloud.pubsub.v1.MessageDispatcher : Sending 281 acks\r\n2018-03-20 14:10:41.729 [Thread-52] DEBUG com.google.cloud.pubsub.v1.MessageDispatcher : Sending 266 acks\r\n2018-03-20 14:10:41.823 [Thread-50] DEBUG com.google.cloud.pubsub.v1.MessageDispatcher : Sending 36 acks\r\n2018-03-20 14:10:41.830 [Thread-42] DEBUG com.google.cloud.pubsub.v1.MessageDispatcher : Sending 75 acks\r\n2018-03-20 14:10:41.832 [Thread-46] DEBUG com.google.cloud.pubsub.v1.MessageDispatcher : Sending 609 acks\r\n2018-03-20 14:10:41.831 [Thread-48] DEBUG com.google.cloud.pubsub.v1.MessageDispatcher : Sending 376 acks\r\n2018-03-20 14:10:41.841 [Thread-44] DEBUG com.google.cloud.pubsub.v1.MessageDispatcher : Sending 186 acks\r\n2018-03-20 14:10:41.841 [Thread-52] DEBUG com.google.cloud.pubsub.v1.MessageDispatcher : Sending 510 acks\r\n2018-03-20 14:10:41.845 [Thread-50] DEBUG com.google.cloud.pubsub.v1.MessageDispatcher : Sending 244 acks\r\n2018-03-20 14:10:41.868 [Thread-42] DEBUG com.google.cloud.pubsub.v1.MessageDispatcher : Sending 412 acks\r\n2018-03-20 14:10:41.887 [Time-limited test] DEBUG com.google.cloud.pubsub.v1.MessageDispatcher : Sending 106 acks\r\n2018-03-20 14:10:41.888 [Time-limited test] DEBUG com.google.cloud.pubsub.v1.MessageDispatcher : Sending 496 acks\r\n2018-03-20 14:10:41.892 [Time-limited test] DEBUG com.google.cloud.pubsub.v1.MessageDispatcher : Sending 379 acks\r\n2018-03-20 14:10:41.894 [Time-limited test] DEBUG com.google.cloud.pubsub.v1.MessageDispatcher : Sending 536 acks\r\n```\r\n\r\nNow, at this point, I can confirm the existence of a backlog by re-running the test but disabling the publisher step. The subscriber should not receive any messages and the test should time out.\r\n\r\nHowever, when I grep for the *Received and Acknowledging * line in my logs, I get a total of *981* rows indicating that those acks that should have been transmitted from the previous run of receiving and acknowledging all 5,000 messages were not received by Cloud Pub/Sub. *981* happens to be the sum of the first three ack batches sent from the *Time-limited test* thread after the subscriber shutdown sequence was invoked.\r\n","author":{"url":"https://github.com/prodonjs","@type":"Person","name":"prodonjs"},"datePublished":"2018-03-20T18:18:26.000Z","interactionStatistic":{"@type":"InteractionCounter","interactionType":"https://schema.org/CommentAction","userInteractionCount":1},"url":"https://github.com/3065/google-cloud-java/issues/3065"}