Repository navigation
PubSub Subscriber will not stop when messages have not been acked or nacked #3752
Description
Issue
When using the Subscriber of the PubSub client, if any message is not ack'ed or nack'ed the subscriber does not shut down anymore.
Why would I want this?
In my case, I want to read PubSub messages using Apache Flink. Flink has a technique to keep itself consistent called checkpointing. Only on every checkpoint will should I ack messages. When a part of Flink crashes/exits/shuts down we want to ignore all messages since the last checkpoint and have PubSub resend them.
This means messages will only be acknowledged on a checkpoint (maybe every 500ms) and when shutting down the last couple of messages won't be acknowledged.
How to reproduce
I've edited the pubsub example in this repo to show what I mean:
public static void main(String... args) throws Exception {
ProjectSubscriptionName subscription = ProjectSubscriptionName.of("bolcom-stg-jey-streaming-886", "streaming-test");
MessageReceiver receiver =
new MessageReceiver() {
@Override
public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) {
System.out.println("Received message: " + message.getData().toStringUtf8());
//DO NOT ack() or nack() and the subscriber will never stop
//consumer.ack();
}
};
Subscriber subscriber = null;
try {
subscriber = Subscriber.newBuilder(subscription, receiver)
.build();
subscriber.addListener(
new Subscriber.Listener() {
@Override
public void failed(Subscriber.State from, Throwable failure) {
// Handle failure. This is called when the Subscriber encountered a fatal error and is shutting down.
System.err.println(failure);
}
},
MoreExecutors.directExecutor());
subscriber.startAsync().awaitRunning();
// In this example, we will pull messages for one minute (60,000ms) then stop.
// In a real application, this sleep-then-stop is not necessary.
// Simply call stopAsync().awaitTerminated() when the server is shutting down, etc.
Thread.sleep(5000);
} finally {
if (subscriber != null) {
System.out.println("Stopping");
subscriber.stopAsync().awaitTerminated();
}
}
}
but basically, all that has changed is I do not ack() or nack() anything.
Question:
The example above won't ever return from the awaitTerminated() call and thus will not exit. Is this behavior by design? Or should I terminate in a different manner?
Activity
chingor13 commentedon Oct 4, 2018
@pongad Is this also related to the fixes you've been working on?
pongad commentedon Oct 5, 2018
This is actually intentional.
awaitTerminatedassumes that you're interested in messages, so it won't return until youackornackall messages. There are a few ways to get what you want.Just don't wait. IIRC, all threads
Subscribercreates internally are daemon threads: if all other threads exit, these threads won't stop the JVM from exiting. If you want to just stop your application this might be an easy way to do it.After pubsub: reject expired and duplicate messages #3743 is merged, we'll correctly give up on messages that take too long to process. If you finish processing messages in 1 min, you can set
setMaxAckExtensionPeriodto 1 min so thatSubscriberwill forget about your messages after about 1 minute. After it forgets,awaitTerminationshould return. Note that after we forget the messages, we won't extend their deadline either, so this option might be a little dangerous.You can use concurrent map to hold
AckConsumer. When youackornack, remove them from the map. When you want to shutdown, nack everything in the map. This might be a little expensive on the CPU.Xeli commentedon Oct 5, 2018
@pongad
Thank you for your comment.
As long as this is intentional it is fine for us to work around this, it just threw me off that awaitTerminated() never actually terminated in our case.
Option 1, unfortunately, does not work for us, because Flink actually removes/unloads classes from the JVM, and those daemon threads start throwing lots of NoClassDefFoundException, so we really want to be sure everything has shutdown before we tell Flink it's oke to unload everything.
Option 3 seems like the way to go, which is fine, we should not buffer too many messages at a given time anyway.
feat: add WRITE_TRUNCATE_DATA as an enum value for write disposition (#…
feat: add WRITE_TRUNCATE_DATA as an enum value for write disposition (#…
fix: use spotify formatter in hermetic build templates (#3752)
chore(main): release 2.56.1 (#3757)