LoomKafkaLibrary

// Using a hypothetical channel API, I assume I have channel `in` with a 4096-wide buffer, 
// and a channel `out` writing to Kafka.

// lookupNodeInRedis and processOnServer have semaphores of, say, 16 and 8, to *globally* limit
// concurrent use of those resources.

HighLevelLib.processInOrder(50, // how many processors
                            in, 
                            HighLevelLib.batchEvery(Duration.ofSeconds(1), out),
         msg ->
                try {
                    var uri = lookupNodeInRedis(msg.record().value()); // blocking
                    var uriDataOffset = new UriDataOffset(uri, msg.record().value(), msg.committableOffset());
                    processOnServer(uriDataOffset.uri, uriDataOffset.data); // blocking
                    return uriDataOffset.offset;
                } catch (Exception e) { ... });