Mrart commented on code in PR #4367:
URL: https://github.com/apache/flink-cdc/pull/4367#discussion_r3593704080
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/external/JdbcSourceFetchTaskContext.java:
##########
@@ -92,43 +92,71 @@ public Object[] getSplitKey(SourceRecord record) {
@Override
public void rewriteOutputBuffer(
Map<Struct, SourceRecord> outputBuffer, SourceRecord changeRecord)
{
- Struct key = (Struct) changeRecord.key();
+ boolean hasPrimaryKey = changeRecord.key() != null;
Struct value = (Struct) changeRecord.value();
if (value != null) {
Envelope.Operation operation =
Envelope.Operation.forCode(value.getString(Envelope.FieldName.OPERATION));
switch (operation) {
case CREATE:
+ if (hasPrimaryKey) {
+ outputBuffer.put(
+ (Struct) changeRecord.key(),
buildReadRecord(changeRecord));
+ } else {
+ Struct after =
value.getStruct(Envelope.FieldName.AFTER);
+ outputBuffer.put(after, buildReadRecord(changeRecord));
+ }
+ break;
case UPDATE:
- Envelope envelope =
Envelope.fromSchema(changeRecord.valueSchema());
- Struct source = value.getStruct(Envelope.FieldName.SOURCE);
- Struct after = value.getStruct(Envelope.FieldName.AFTER);
- Instant fetchTs =
- Instant.ofEpochMilli((Long)
source.get(Envelope.FieldName.TIMESTAMP));
- SourceRecord record =
- new SourceRecord(
- changeRecord.sourcePartition(),
- changeRecord.sourceOffset(),
- changeRecord.topic(),
- changeRecord.kafkaPartition(),
- changeRecord.keySchema(),
- changeRecord.key(),
- changeRecord.valueSchema(),
- envelope.read(after, source, fetchTs));
- outputBuffer.put(key, record);
+ if (hasPrimaryKey) {
+ outputBuffer.put(
+ (Struct) changeRecord.key(),
buildReadRecord(changeRecord));
+ } else {
+ // For no-PK table: remove the before image, insert
the after image
Review Comment:
Is there something wrong with the log message?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
