Computation without input assignement should discard timer execution

When a computation has an input stream, if the computation has no assigned partition (after a rebalancing) the timer should not be executed.

The rational for this is to be able to create unique producer.

We can already have a unique consumer, just creating a stream with one partition, the computation can run on multiple nodes but only one will be elected by Kafka to process the records.

Now if we want to create a unique producer or source computation, we can use the same principe reading from an empty stream with a single partition and produce message using the timer callback. But this means that only the computation that have the assigned partition run the timer.

This could be as simple as:

--- a/nuxeo-runtime/nuxeo-stream/src/main/java/org/nuxeo/lib/stream/computation/log/ComputationRunner.java
+++ b/nuxeo-runtime/nuxeo-stream/src/main/java/org/nuxeo/lib/stream/computation/log/ComputationRunner.java
@@ -209,6 +209,10 @@ public class ComputationRunner implements Runnable, RebalanceListener {
         if (timers.isEmpty()) {
             return;
         }
+        if (tailer != null && tailer.assignments().isEmpty()) {
+            // needed to ensure single source across multiple nodes
+            return;
+        }
         long now = System.currentTimeMillis();
         final boolean[] timerUpdate = { f