How to check when polling stopped

I have a message stream, where messages comes which I need to process and then store them in database. In Java, I've written polling code which polls stream and consumes messages every 20 seconds.

This is done inside an infinite for-loop, like below:

for (;;) {
    try{
        //1. Logic for polling.
        //2. Logic for processing the message.
        //3. Logic for storing the message in database.
        Thread.sleep(20000 - <time taken for above 3 steps >); 
    } catch(Exception E){
        //4. Exception handling.
    }
}

This logic runs as expected and the stream is polled, but once in a while it hits an exception or something goes wrong and polling stops.

I want to have a mechanism, that as soon as polling stopped, let's say this for loop is not running for 60 seconds, I should receive a mail or ping.

How can I invoke a method if this for loop is not running for 60 seconds?

I am thinking like, each for-loop execution will ping a heartbeat, and when that heartbeat pinging not received from for-loop then a mail sending is invoked.

There are two different reasons why polling stops making progress, and each needs a different approach:

If the logic throws a Throwable other than an Exception, for instance an Error, the catch does not match, and execution will leave the for-loop, and likely reach the thread's UncaughtExceptionHandler, the default implementation of which logs the exception to System.err and terminates the thread. To prevent this, you should catch Throwable rather than Exception.

The second possibility is that some step in your logic doesn't terminate, for instance due to an infinite loop, a deadlock, waiting for I/O operations, or whatever. In this case, you'll want to take a thread dump to see where the thread is stuck. You can automate this as follows:

class Watchdog {
	final Duration gracePeriod;
	final Thread watchedThread;
	volatile Instant lastProgress;

	public Watchdog(Duration gracePeriod) {
		this.gracePeriod = gracePeriod;
		watchedThread = Thread.currentThread();
		everythingIsFine();
		
		var t = new Thread(this::keepWatch);
		t.setDaemon(true);
		t.start();
	}
	
	public void everythingIsFine() {
		lastProgress = Instant.now();
	}
	
	void keepWatch() {
		while (true) {
			var silence = Duration.between(lastProgress, Instant.now());
			if (silence.compareTo(gracePeriod) > 0) {
				System.err.println("Watchdog hasn't seen any progress for " + silence.toSeconds() + " seconds. The watched thread is currently at:");
				for (var element : watchedThread.getStackTrace()) {
					System.err.println("\tat " + element);
				}
			}
			try {
				Thread.sleep(gracePeriod);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}
}

while you can use as follows:

public class Test {
	void step() throws Exception {
		System.in.read();
	}
	
	void job() {
		var snoopy = new Watchdog(Duration.ofSeconds(2));
		for (;;) {
			try {
				step();
				snoopy.everythingIsFine();
				Thread.sleep(1000);
			} catch (Throwable t) {
				System.err.println(t);
			}
		}
	}
	
	public static void main(String[] args) throws Exception {
		new Test().job();
	}
}

once the grace period elapses, the WatchDog will print something like:

<pre>
Watchdog hasn't seen any progress for 2 seconds. The watched thread is currently at:
at java.base/java.io.FileInputStream.readBytes(Native Method)
at java.base/java.io.FileInputStream.read(FileInputStream.java:293)
at java.base/java.io.BufferedInputStream.fill(BufferedInputStream.java:255)
at java.base/java.io.BufferedInputStream.implRead(BufferedInputStream.java:289)
at java.base/java.io.BufferedInputStream.read(BufferedInputStream.java:276)
at stackoverflow.Test.step(Test.java:48)
at stackoverflow.Test.job(Test.java:55)
at stackoverflow.Test.main(Test.java:65)
</pre>