-
Notifications
You must be signed in to change notification settings - Fork 8
Open
Description
class SinkConfig
static _messageToRecord(message) {
//check if a converter has already turned this message into a record
if (message && typeof message.value === "object" &&
message instanceof SinkRecord) {
return message;
}
try {
const record = new SinkRecord();
record.kafkaOffset = message.offset;
record.key = message.key;
record.partition = message.partition;
record.keySchema = message.value.keySchema;
record.timestamp = message.value.timestamp;
record.value = message.value.value;
record.valueSchema = message.value.valueSchema;
return record;
} catch (error) {
debug("Failed to turn message into sink record.", error.message);
super.emit("error", "Failed to turn message into SinkRecord: " + error.message);
return message;
}
}
Metadata
Metadata
Assignees
Labels
No labels