我需要能够标记AVRO模式中的某些字段,以便在序列化时对其进行加密。
逻辑类型允许标记这些字段,并与自定义转换一起使用,应该可以让AVRO无缝地将它们加密。
我曾经遇到过一些问题,无法找到如何在AVRO中定义和使用新的逻辑类型的文档(avro_1.8.2#逻辑类型)。
因此,我决定在答案中分享我发现的内容,以使其他人能够更轻松地了解并获取反馈,以防我做错了什么。
我需要能够标记AVRO模式中的某些字段,以便在序列化时对其进行加密。
逻辑类型允许标记这些字段,并与自定义转换一起使用,应该可以让AVRO无缝地将它们加密。
我曾经遇到过一些问题,无法找到如何在AVRO中定义和使用新的逻辑类型的文档(avro_1.8.2#逻辑类型)。
因此,我决定在答案中分享我发现的内容,以使其他人能够更轻松地了解并获取反馈,以防我做错了什么。
首先,我定义了一个逻辑类型:
public class EncryptedLogicalType extends LogicalType {
//The key to use as a reference to the type
public static final String ENCRYPTED_LOGICAL_TYPE_NAME = "encrypted";
EncryptedLogicalType() {
super(ENCRYPTED_LOGICAL_TYPE_NAME);
}
@Override
public void validate(Schema schema) {
super.validate(schema);
if (schema.getType() != Schema.Type.BYTES) {
throw new IllegalArgumentException(
"Logical type 'encrypted' must be backed by bytes");
}
}
}
然后是一次新的转换:
public class EncryptedConversion extends Conversion<ByteBuffer> {
// Construct a unique instance for all the conversion. This have to be changed in case the conversion
// needs some runtime information (e.g.: an encryption key / a tenant_ID). If so, the get() method should
// return the appropriate conversion per key.
private static final EncryptedConversion INSTANCE = new EncryptedConversion();
public static final EncryptedConversion get(){ return INSTANCE; }
private EncryptedConversion(){ super(); }
//This conversion operates on ByteBuffer and returns ByteBuffer
@Override
public Class<ByteBuffer> getConvertedType() { return ByteBuffer.class; }
@Override
public String getLogicalTypeName() { return EncryptedLogicalType.ENCRYPTED_LOGICAL_TYPE_NAME; }
// fromBytes and toBytes have to be overridden as this conversion works on bytes. Other may need to be
// overridden. The types supported need to be updated also in EncryptedLogicalType#validate(Schema schema)
@Override
public ByteBuffer fromBytes(ByteBuffer value, Schema schema, LogicalType type) {
encryptedValue = __encryptionLogic__(value);
return encryptedValue;
}
@Override
public ByteBuffer toBytes(ByteBuffer value, Schema schema, LogicalType type) {
decryptedValue = __decryptionLogic__(value);
return decryptedValue;
}
}
.avsc模式文件将类似于:
{
"name": “MyMessageWithEncryptedField”,
"type": "record",
"fields": [
{"name": "payload","type" : {"type" : "bytes","logicalType" : "encrypted"}},
...
在从模式文件生成的MyMessageWithEncryptedField.java
类中,最后我添加了返回转换的方法:
@Override
public Conversion<?> getConversion(int fieldIndex) {
// This allow us to have a more flexible conversion retrieval, so we don't have to code it per field.
Schema fieldSchema = SCHEMA$.getFields().get(fieldIndex).schema();
if ((fieldSchema.getLogicalType() != null)
&& (fieldSchema.getLogicalType().getName() == EncryptedLogicalType.ENCRYPTED_LOGICAL_TYPE_NAME)){
// here we could pass to the get() method a runtime information, e.g.: a tenantId that can be found in the data structure.
return EncryptedConversion.get();
}
return null;
}
为了使其运行,我仍然需要在运行时注册类型:
LogicalTypes.register(EncryptedLogicalType.ENCRYPTED_LOGICAL_TYPE_NAME, new LogicalTypes.LogicalTypeFactory() {
private final LogicalType encryptedLogicalType = new EncryptedLogicalType();
@Override
public LogicalType fromSchema(Schema schema) {
return encryptedLogicalType;
}
});
注意事项:
MyMessageWithEncryptedField.java
)的静态块内。