class KafkaProducerRecordBuilder {

    static List<KafkaProducerRecord<String, String>> buildRecords(String topic, Buffer payload) throws ValidationException {
        List<KafkaProducerRecord<String, String>> kafkaProducerRecords = new ArrayList<>();
        JsonObject payloadObj;
        try {
            payloadObj = new JsonObject(payload);
        } catch (DecodeException de){
            throw new ValidationException("Error while parsing payload");
        }

        JsonArray recordsArray;
        try {
            recordsArray = payloadObj.getJsonArray(RECORDS);
            if(recordsArray == null){
                throw new ValidationException("Missing 'records' array");
            }
            for (int i = 0; i < recordsArray.size(); i++) {
                kafkaProducerRecords.add(fromRecordJsonObject(topic, recordsArray.getJsonObject(i)));
            }
        } catch (ClassCastException cce) {
            throw new ValidationException("Property '"+RECORDS+"' must be of type JsonArray holding JsonObject objects");
        }

        return kafkaProducerRecords;
    }

}