[KafkaConnect] Fix RecordConverter for UUID and Fixed Types

[singhpk234]

About the change

The UUID type in the parquet writer expects ByteBuffer rather than UUID otherwise writer fails with :

class java.util.UUID cannot be cast to class [B (java.util.UUID and [B are in module java.base of loader 'bootstrap')

The FixedLength would need byteArray rather than ByteBuffer otherwise one get this error

class java.nio.HeapByteBuffer cannot be cast to class [B (java.nio.HeapByteBuffer and [B are in module java.base of loader 'bootstrap')

Testing

Added new tests

cc @bryanck

[bryanck]

I was hoping we'd keep this test specific to the conversion functions, and keep writer tests separate. Do you have thoughts on that?

[singhpk234]

Was thinking in lines of conversion functions are no longer format agnostic as we are adding format info into deciding the record conversion, hence though it would be fair to test this E2E here.

please let me know your thoughts considering above.

[jbonofre]

Maybe we can create a dedicated test class for writer ?

[RussellSpitzer]

Shouldn't we have this parameterized by file type? The tests here make sense to me but I am only looking at this module for the first time for this PR.

[RussellSpitzer]

Ah iI see, the format only comes into play for UUID so the other parameterizations are essentially no-ops. Perhaps we just need one specialized test then ' testParquetUUIDSerialization"

[singhpk234]

Sure, added this test, Thanks for suggesting this !
I also do think we need an E2E with write I can take this test as a followup to this pr as it would require refactoring of the writer tests.

[jbonofre]

This change looks good, I'm just wondering about the test. I would have kept the original test and create a new one dedicated for the writer.

[RussellSpitzer]

This looks good to me on the fix side but I agree with the others than we need to adjust the tests to be a bit more specific to this fix I think?

[RussellSpitzer]

Thanks @singhpk234 for the PR and @jbonofre, @bryanck and @ajantha-bhat For Review!

[Gezi-lzq]

When writing UUIDs, should we handle the conversion directly within BaseParquetWriter, by modifying BaseParquetWriter#primitive to check if the LogicalTypeAnnotation is a UUID and then use a UUIDWriter for writing it, instead of performing the conversion based on the file type before writing?

@Override
public ParquetValueWriter<?> primitive(PrimitiveType primitive) {
    ....
    switch (primitive.getPrimitiveTypeName()) {
        case FIXED_LEN_BYTE_ARRAY:
            if (LogicalTypeAnnotation.uuidType().equals(primitive.getLogicalTypeAnnotation())) {
                return new UUIDWriter(desc);
            }
            return new FixedWriter(desc);

            ...
    }
}

private static class UUIDWriter extends ParquetValueWriters.PrimitiveWriter<UUID> {
    private UUIDWriter(ColumnDescriptor desc) {
        super(desc);
    }

    @Override
    public void write(int repetitionLevel, UUID value) {
        column.writeBinary(repetitionLevel, Binary.fromReusedByteArray(UUIDUtil.convert(value)));
    }
}

Similar to the approach taken in #7399 in Apache Iceberg.
@singhpk234 @bryanck @openinx @RussellSpitzer