Time-limit the consumer using the latch's await method; don't attempt to figure it out based on messages arriving.
Merge bug26379