BugFix: Fixed an issue where JDBC OceanBaseSink might cause data loss in non-primary key tables

[yuanoOo]

Summary

fix #109

Solution Description

[whhe]

What's it for? The process should directly go to orElse if the record is from a non-pk table,

[yuanoOo]

flink-connector-oceanbase/flink-connector-oceanbase-base/src/main/java/com/oceanbase/connector/flink/sink/OceanBaseWriter.java

Lines 132 to 144 in f7f175f

	Object key = keyExtractor.extract(dataChangeRecord);
	if (key == null) {
	synchronized (buffer) {
	buffer.computeIfAbsent(record.getTableId().identifier(), k -> new ArrayList<>())
	.add(dataChangeRecord);
	}
	} else {
	synchronized (reducedBuffer) {
	reducedBuffer
	.computeIfAbsent(record.getTableId().identifier(), k -> new HashMap<>())
	.put(key, dataChangeRecord);
	}
	}

• If record.getTable().getKey() is an empty but not null primary key List, this KeyExtractor.simple() will return an empty DataChangeRecord object but not NULL
• Then line 133 of the code block above will return false, causing the non-primary key table to be mistakenly treated as the primary key table.

        static KeyExtractor simple() {
            return record ->
                    // when record.getTable().getKey() is empty List
                    Optional.ofNullable(record.getTable().getKey())
                            .map(
                                    keys ->
                                            new DataChangeRecordData(
                                                    keys.stream()
                                                            .map(record::getFieldValue)
                                                            .toArray()))
                            .orElse(null);
        }