Kafka Connect是否支持枚举?

问题描述：

Kafka Connect是否支持枚举字段?如果没有，通常的解决方法是什么?我在这里查看Kafka 2.6.0 ConnectSchema API:

Are enum fields supported in Kafka Connect? If not, what is the usual workaround? I'm looking at the Kafka 2.6.0 ConnectSchema API here: https://kafka.apache.org/26/javadoc/org/apache/kafka/connect/data/ConnectSchema.html

我正在尝试通过使用Confluent Schema注册表(使用AVRO)来遵循最佳实践，但是似乎无法让我的自定义源连接器生成包含枚举以匹配现有模式的模式(输出主题还有其他除连接器外的其他生产商).一种解决方法是仅使用字符串，但这会破坏模式的整个重点，不是吗?

I'm trying to follow best practice by using the Confluent Schema registry (with AVRO), but can't seem to get my custom source connector to generate a schema containing enums to match an existing schema (the output topic has other producers besides the connector). A work-around would be to simply use strings, but that undermines the whole point of a schema, doesn't it?

答

没有一种通用的方法，但是对于AVRO，您可以使用特定于转换器的特殊配置，然后还必须提供特殊提示通过枚举字段的Schema属性.我能够使用自定义的Connect Transform提供提示.

There is not an approach that works in general, but you can use a converter specific special configuration in the case of AVRO, and then you must also provide special hints via the enum field Schema properties. I was able to provide the hints using a custom Connect Transform.

使用以下方法配置Connect转换器:

Configure the Connect converter with:

"value.converter": "io.confluent.connect.avro.AvroConverter",
"value.converter.schema.registry.url": "http://registry:8081",
"value.converter.enhanced.avro.schema.support":  true,
"value.converter.connect.meta.data": false,
"transforms": "alarms",
"transforms.alarms.type": "org.jlab.kafka.connect.transforms.EpicsToAlarm$Value"

然后自定义转换包含:

    final Schema prioritySchema = SchemaBuilder
            .string()
            .doc("Alarm severity organized as a way for operators to prioritize which alarms to take action on first")
            .parameter("io.confluent.connect.avro.enum.doc.AlarmPriority", "Enumeration of possible alarm priorities")
            .parameter("io.confluent.connect.avro.Enum", "org.jlab.kafka.alarms.AlarmPriority")
            .parameter("io.confluent.connect.avro.Enum.1", "P1_LIFE")
            .parameter("io.confluent.connect.avro.Enum.2", "P2_PROPERTY")
            .parameter("io.confluent.connect.avro.Enum.3", "P3_PRODUCTIVITY")
            .parameter("io.confluent.connect.avro.Enum.4", "P4_DIAGNOSTIC")
            .build();

完整转换源

Kafka Connect是否支持枚举?

相关推荐