LoomKafka

// 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.

var processingQueue = new BufferedChannel(50); // this buffer bounds the number of processing threads

// Consume and process input
Thread.startVirtualThread(() -> {
    try (var e = Executors.newUnboundedVirtualThreadExecutor()) {
        while (true) {
            var msg = in.receive(); // for each incoming message we start a processing thread
            var future = e.submit(() -> {
                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) { ... }
            });
            processingQueue.send(future); // we block here if there are too many processing threads
        }
    }
});

// Collect and batch output
Thread.startVirtualThread(() -> {
    while (true) {
        var batch = new ArrayList<Result>();

        try (var e = Executors.newUnboundedVirtualThreadExecutor()
                              .withDeadline(Instant.now().plusSeconds(1))) {
            e.submit(() ->
                while (true) {
                    window.add(processingQueue.map(Future::get).receive()); // WARNING: map must be atomic
                }
            });
        } catch (Exception e) { ... }

        out.send(batch); 
    }
}).join();