kc.java

class CamelSource {
    int maxElements;
    int timeout;
    TimeUnit timeoutUnit;
    List<SourceRecord> records;
    BlockingQueue<SourceRecord> queue;

    @Override
    public void start(Map<jString, String> props) {
        maxElements = props.get();
        timeout = props.get()
        timeoutUnit = props.get()

        records = new ArrayList<>(maxElements);
        queue = new ArrayBlockingQueue<>(maxElements);

        context.registry().bind("kafka-connect-record", new KafkaConnectDataFormat())
        context.addRoute(new RouteBuilder() {
            @Override
            public void configure() throws Exception {
                from("kamelet:staceppa")
                    .marshal("kafka-connect-record")
                    .process(queue::offer)
            }
        });
    }

    @Override
    public List<SourceRecord> poll() { 
        final Record record = queue.poll(timeout, timeoutUnit);

        if (record != null) {
            records.clear();
            records.add(record);

            queue.drain(records, maxElements - records.size());

            return records;
        } else {
            retun null;
        }
    }
}