home/schema-registry-transfer-smt
2
0
Fork 0
mirror of https://github.com/wahyd4/schema-registry-transfer-smt.git synced 2026-08-09 04:25:59 +10:00
Code Issues Packages Projects Releases Wiki Activity

- handle case when schema info is present but value is null

Browse Source 
- not skipping record in that case but omitting schema re-registration
This commit is contained in:
Tayfun Wiechert
2019-03-15 02:23:25 -05:00
committed by Jordan Moore
parent 443ae8d77c
commit a40d8a97dc
1 changed files with 10 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/cricket/jmoore/kafka/connect/transforms/SchemaRegistryTransfer.java
+10 -8
View File
@@ -124,16 +124,18 @@ public class SchemaRegistryTransfer<R extends ConnectRecord<R>> implements Trans
Object updatedValue;
Optional<Integer> destValueSchemaId;
if (value == null) {
throw new ConnectException("Unable to extract schema information from null record value.");
}
if ((valueSchema != null && valueSchema.type() == Schema.BYTES_SCHEMA.type()) ||
value instanceof byte[]) {
ByteBuffer b = ByteBuffer.wrap((byte[]) value);
destValueSchemaId = copySchema(b, topic, false);
b.putInt(1, destValueSchemaId.orElseThrow(()
-> new ConnectException("Transform failed. Unable to update record schema id. (isKey=false)")));
updatedValue = b.array();
if(value!=null) {
ByteBuffer b = ByteBuffer.wrap((byte[]) value);
destValueSchemaId = copySchema(b, topic, false);
b.putInt(1, destValueSchemaId.orElseThrow(()
-> new ConnectException("Transform failed. Unable to update record schema id. (isKey=false)")));
updatedValue = b.array();
} else {
log.trace("Cannot extract schema details from null-value record.");
updatedValue = value;
}
} else {
throw new ConnectException("Transform failed. Record value does not have a byte[] schema.");
}