nleena123 commented on issue #5540:
URL: https://github.com/apache/hudi/issues/5540#issuecomment-1128769624
We are adding new field CDC_TS to Kafka data through code which present in
DF3HoodieDeltaStreamer.java program.
**Code :-**
static GenericRecord getDataWithCDC_TS(GenericRecord record, String ts) {
Schema base = record.getSchema();
List<Schema.Field> baseFields = base.getFields().stream()
.map(field -> {
Schema.Field f = new
Schema.Field(field.name(), field.schema(), field.doc(), field.defaultVal());
field.aliases().forEach(f::addAlias);
return f;
})
.collect(Collectors.toList());
List<Schema.Field> newFields = new ArrayList<>(baseFields);
ArrayList<Schema> optionalString = new ArrayList<>();
optionalString.add(Schema.create(Schema.Type.NULL));
optionalString.add(Schema.create(Schema.Type.STRING));
//JsonNode nullDefault = JsonNodeFactory.instance.nullNode();
//Schema.Field f = new Schema.Field(CDC_TS,
Schema.createUnion(optionalString), CDC_TS, nullDefault);
Schema.Field f = new Schema.Field(CDC_TS,
Schema.createUnion(optionalString), CDC_TS, null);
if (!baseFields.contains(f))
newFields.add(f);
Schema newSchema = Schema.createRecord(
base.getName(),
"New schema by adding a new field,CDC_TS",
base.getNamespace(), base.isError(),
newFields);
LOG.info("newSchema is:" + newSchema);
GenericRecord newRecord = new GenericData.Record(newSchema);
base.getFields().stream().forEach(field ->
newRecord.put(field.name(), record.get(field.name())));
newRecord.put(CDC_TS, ts);
return newRecord;
}
i have attached Latest log file , please have look on it.
[log4j-active_latest.txt](https://github.com/apache/hudi/files/8707908/log4j-active_latest.txt)
--
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]
