Kafka Streams JDBC源长期不兼容

Question

问题：在设置一个 Kafka 管道后，该管道使用 Kafka Connect JDBC 源和 Avro 序列化程序和反序列化程序将数据拉入，一旦我尝试使用 Kafka Streams Java 应用程序将该数据读入 KStream，就会收到以下错误。

org.apache.kafka命令错误。SerializationException:LongDeserializer接收的数据大小不是8

我试图尽可能地遵循现有的示例，但有些事情没有意义。我将在下面包含所有代码/附加信息，但我有几个问题…

>

我目前在理解上的最大差距之一是Avro记录的“KEY”使用了什么？（在运行时）在我身上出错的行与我告诉KStream键是LONG的事实有关，然而当检索Avro记录时，长度小于8（Long类型的预期长度）。
当我设置我的JDBC源时，那里没有任何东西可以标识键是什么——我在留档中也没有看到任何东西会让我相信我可以指定键，尽管我已经尝试过了：

curl -X POST \
  -H "Content-Type: application/json" \
  --data 'see next code block for formatted data'  \
http://localhost:8083/connectors

// This is the data chunk used above but in a string - broke it apart for readability here
{
    "name": "source-jdbc-ldw_applications",
    "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
        "tasks.max": 1,
        "connection.url": "jdbc:sqlserver://dbserver;databaseName=dbname;user=kafkareader;password=kafkareader;",
        "mode": "incrementing",
        "incrementing.column.name": "ApplicationID",
        "topic.prefix": "source-jdbc-",
        "poll.interval.ms": 30000,
        "table.whitelist": "LDW_Applications",
        "transforms": "setSchema",
        "transforms.setSchema.type": "org.apache.kafka.connect.transforms.SetSchemaMetadata$Value",
        "transforms.setSchema.schema.name": "com.mycompany.avro.Application",
        "transforms.setSchema.schema.version": "1"
    }
}

通过上述操作，我得到了运行报告的以下模式：

curl http://localhost:8081/subjects/source-jdbc-LDW_Applications-value/versions/1 |jq

这是它的输出：

{
    "subject": "source-jdbc-LDW_Applications-value",
    "version": 1,
    "id": 9,
    "schema": "{\"type\":\"record\",\"name\":\"Application\",\"namespace\":\"com.baydynamics.avro\",\"fields\":[{\"name\":\"ApplicationID\",\"type\":\"long\"},{\"name\":\"Name\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"Description\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"Group\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"OwnerUserID\",\"type\":[\"null\",\"long\"],\"default\":null},{\"name\":\"RiskScore\",\"type\":[\"null\",{\"type\":\"int\",\"connect.type\":\"int16\"}],\"default\":null},{\"name\":\"RiskRating\",\"type\":[\"null\",\"string\"],\"default\":null},{\"name\":\"ServiceLevelTierID\",\"type\":[\"null\",\"int\"],\"default\":null},{\"name\":\"LossPotentialID\",\"type\":[\"null\",\"int\"],\"default\":null},{\"name\":\"ConfidentialityRequirementID\",\"type\":[\"null\",\"int\"],\"default\":null},{\"name\":\"IntegrityRequirementID\",\"type\":[\"null\",\"int\"],\"default\":null},{\"name\":\"AvailabilityRequirementID\",\"type\":[\"null\",\"int\"],\"default\":null},{\"name\":\"ApplicationCategoryID\",\"type\":[\"null\",\"long\"],\"default\":null}],\"connect.version\":1,\"connect.name\":\"com.baydynamics.avro.Application\"}"
}

为了让这个模式看起来更漂亮一些:

{
"type":"record",
"name":"Application",
"namespace":"com.baydynamics.avro",
"fields":[
    {
        "name":"ApplicationID",
        "type":"long"
    },
    {
        "name":"Name",
        "type":[
            "null",
            "string"
        ],
        "default":null
    },
    {
        "name":"Description",
        "type":[
            "null",
            "string"
        ],
        "default":null
    },
    {
        "name":"Group",
        "type":[
            "null",
            "string"
        ],
        "default":null
    },
    {
        "name":"OwnerUserID",
        "type":[
            "null",
            "long"
        ],
        "default":null
    },
    {
        "name":"RiskScore",
        "type":[
            "null",
            {
            "type":"int",
            "connect.type":"int16"
            }
        ],
        "default":null
    },
    {
        "name":"RiskRating",
        "type":[
            "null",
            "string"
        ],
        "default":null
    },
    {
        "name":"ServiceLevelTierID",
        "type":[
            "null",
            "int"
        ],
        "default":null
    },
    {
        "name":"LossPotentialID",
        "type":[
            "null",
            "int"
        ],
        "default":null
    },
    {
        "name":"ConfidentialityRequirementID",
        "type":[
            "null",
            "int"
        ],
        "default":null
    },
    {
        "name":"IntegrityRequirementID",
        "type":[
            "null",
            "int"
        ],
        "default":null
    },
    {
        "name":"AvailabilityRequirementID",
        "type":[
            "null",
            "int"
        ],
        "default":null
    },
    {
        "name":"ApplicationCategoryID",
        "type":[
            "null",
            "long"
        ],
        "default":null
    }
],
"connect.version":1,
"connect.name":"com.baydynamics.avro.Application"
}

所以，再一次，我没有看到任何东西表明上面的任何特定字段将是记录的关键。

然后我进入Kafka Streams，我试着把这些数据放入KStream...它爆炸了...

final KStream<Long, Application> applicationStream = builder.stream(Serdes.Long(), applicationSerde, VULNERABILITY_TOPIC);

所以，事情是这样的，因为我知道存储在幕后的数据是SQL Server中的BIGINT，并且映射到Java中的LONG，所以我将KStream的密钥类型设置为Long，然后我使用Serdes.Long（）反序列化器作为KStream构建器的参数。

调试时，我看到原始记录的长度为 7，这就是它引发错误的原因。显然 Avro 以一种更好地压缩的方式序列化东西？我不知道。无论如何，问题是我什至不知道它认为它实际上使用的密钥是什么？！所以谁知道 - 也许我对 Long 的假设是不正确的，因为它实际上并没有使用 ApplicationID 作为键？为什么我会认为是？！

任何关于这方面的帮助将不胜感激。我知道上面有很多信息，但简单来说..

1. 使用JDBC Kafka连接将数据推送到主题
2. 数据正在进入主题-我可以通过控制台看到它
3. 试图将这些数据推送到流中，这样我就可以用这些数据做一些很棒的事情，但由于Serdes与Avro Record不兼容，尝试填充流时会失败

Answer 1

ConFluent JDBC源连接器不会生成带有密钥的记录。已经记录了添加此支持的功能请求。

同时，您可以使用单个消息转换从值中提取一些字段，从而实质上创建密钥。内置的ValueToKey转换就是这样做的。这篇博客文章有一个SMT的例子。