class RecordServiceImpl implements RecordService, Serializable {

    @Override
    public <T> T visit(final RecordVisitor<T> visitor, final Record record) {
        final AtomicReference<T> out = new AtomicReference<>();
        record.getSchema().getEntries().forEach(entry -> {
            switch (entry.getType()) {
            case INT:
                visitor.onInt(entry, record.getOptionalInt(entry.getName()));
                break;
            case LONG:
                visitor.onLong(entry, record.getOptionalLong(entry.getName()));
                break;
            case FLOAT:
                visitor.onFloat(entry, record.getOptionalFloat(entry.getName()));
                break;
            case DOUBLE:
                visitor.onDouble(entry, record.getOptionalDouble(entry.getName()));
                break;
            case BOOLEAN:
                visitor.onBoolean(entry, record.getOptionalBoolean(entry.getName()));
                break;
            case STRING:
                visitor.onString(entry, record.getOptionalString(entry.getName()));
                break;
            case DATETIME:
                visitor.onDatetime(entry, record.getOptionalDateTime(entry.getName()));
                break;
            case BYTES:
                visitor.onBytes(entry, record.getOptionalBytes(entry.getName()));
                break;
            case RECORD:
                final Optional<Record> optionalRecord = record.getOptionalRecord(entry.getName());
                final RecordVisitor<T> recordVisitor = visitor.onRecord(entry, optionalRecord);
                optionalRecord.ifPresent(r -> {
                    final T visited = visit(recordVisitor, r);
                    if (visited != null) {
                        final T current = out.get();
                        out.set(current == null ? visited : visitor.apply(current, visited));
                    }
                });
                break;
            case ARRAY:
                final Schema schema = entry.getElementSchema();
                switch (schema.getType()) {
                case INT:
                    visitor.onIntArray(entry, record.getOptionalArray(int.class, entry.getName()));
                    break;
                case LONG:
                    visitor.onLongArray(entry, record.getOptionalArray(long.class, entry.getName()));
                    break;
                case FLOAT:
                    visitor.onFloatArray(entry, record.getOptionalArray(float.class, entry.getName()));
                    break;
                case DOUBLE:
                    visitor.onDoubleArray(entry, record.getOptionalArray(double.class, entry.getName()));
                    break;
                case BOOLEAN:
                    visitor.onBooleanArray(entry, record.getOptionalArray(boolean.class, entry.getName()));
                    break;
                case STRING:
                    visitor.onStringArray(entry, record.getOptionalArray(String.class, entry.getName()));
                    break;
                case DATETIME:
                    visitor.onDatetimeArray(entry, record.getOptionalArray(ZonedDateTime.class, entry.getName()));
                    break;
                case BYTES:
                    visitor.onBytesArray(entry, record.getOptionalArray(byte[].class, entry.getName()));
                    break;
                case RECORD:
                    final Optional<Collection<Record>> array = record.getOptionalArray(Record.class, entry.getName());
                    final RecordVisitor<T> recordArrayVisitor = visitor.onRecordArray(entry, array);
                    array.ifPresent(a -> a.forEach(r -> {
                        final T visited = visit(recordArrayVisitor, r);
                        if (visited != null) {
                            final T current = out.get();
                            out.set(current == null ? visited : visitor.apply(current, visited));
                        }
                    }));
                    break;
                // array of array is not yet supported!
                default:
                    throw new IllegalStateException("Unsupported entry type: " + entry);
                }
                break;
            default:
                throw new IllegalStateException("Unsupported entry type: " + entry);
            }
        });
        final T value = out.get();
        final T visited = visitor.get();
        if (value != null) {
            return visitor.apply(value, visited);
        }
        return visited;
    }

}