Kafka Connect SMT ApplyWithSchema 需要结构错误答案

【问题标题】：Kafka Connect SMT ApplyWithSchema requires struct errorKafka Connect SMT ApplyWithSchema 需要结构错误
【发布时间】：2020-11-28 08:41:04
【问题描述】：

我已经从 confluent https://github.com/confluentinc/kafka-connect-insert-uuid 部署了一个示例，用于添加简单的 UUID 字段，但我收到一个错误，它需要结构。我在 Debezium MySQLConnector 中应用这个

  Only Struct objects supported for [adding UUID to record], found: 
  java.lang.String\n\tat org.apache.kafka.connect.transforms.util.Requirements.requireStruct(Requirements.java:52)

什么是只返回记录的极简 applyWithSchema 方法？我正在尝试调试并需要一个没有任何错误的 HelloWorld SMT，必须应用包括 applyWithSchema 在内的方法

我认为这对于应用程序来说可能是最简单的，但需要 applyWithSchema

Override
public R apply(R record) {
    
    return record.newRecord(
            record.topic(), record.kafkaPartition(),
            record.keySchema(), record.key(),
            record.valueSchema(), record.value(),
            record.timestamp()
    );
}

Override
public R applyWithSchema(R record) {
    // what is minimal transform here??
}

我现在只需要这些函数运行而不会出错，因为我只对 record.headers().add() 进行了更改。

这是给出错误的 applyWithSchema 方法：

private R applyWithSchema(R record) {
    // FAILS HERE!
    final Struct value = requireStruct(operatingValue(record), PURPOSE);

    Schema updatedSchema = schemaUpdateCache.get(value.schema());
    if(updatedSchema == null) {
        updatedSchema = makeUpdatedSchema(value.schema()); 

        final Struct updatedValue = new Struct(updatedSchema);

        for (Field field : value.schema().fields()) {
         // updatedValue.put(field.name(), value.get(field));
        }

        //updatedValue.put(fieldName, getRandomUuid());

        return newRecord(record, updatedSchema, updatedValue);
    }

【问题讨论】：

您可以使用Hoist 转换来获取结构。

标签： java apache-kafka apache-kafka-connect

【解决方案1】：

主要问题在于您的连接器转换配置顺序，当您定义连接器时，连接器将按照同步顺序应用这些 SMT 操作。如果您期望自定义 SMT 的结构或本机信息，请确保以前的 SMT 不会改变自然状态。

"transforms":"SMT1, SMT2, CustomSMT3",

【讨论】：