Kafka Persistence State-Store 不适用于 python 和 java 的组合使用答案

【问题标题】：Kafka Persistence State-Store Not Worked with combined use of python and javaKafka Persistence State-Store 不适用于 python 和 java 的组合使用
【发布时间】：2020-06-13 05:51:23
【问题描述】：

今天我在 Kafka 国营商店里发现了很奇怪的东西，我用谷歌搜索了很多，但没有找到行为的原因。

考虑以下用 java 编写的状态存储：

private KeyValueStore<String, GenericRecord> userIdToUserRecord;

有两个处理器正在使用这个状态存储。

  topology.addStateStore(userIdToUserRecord, ALERT_PROCESSOR_NAME, USER_SETTING_PROCESSOR_NAME)

USER_SETTING_PROCESSOR_NAME会将数据放入状态存储

userIdToUserRecord.put("user-12345", record);

ALERT_PROCESSOR_NAME将从状态存储中获取数据

userIdToUserRecord.get("user-12345");

向 UserSettingProcessor 添加源

userSettingTopicName = user-setting-topic;    
topology.addSource(sourceName, userSettingTopicName)
                    .addProcessor(processorName, UserSettingProcessor::new, sourceName);

向 AlertEngineProcessor 添加源

alertTopicName = alert-topic;
topology.addSource(sourceName, alertTopicName)
                    .addProcessor(processorName, AlertEngineProcessor::new, sourceName);

案例 1： 在java中使用Kafka生产生产记录
首先使用 java 为主题 user-setting-topic 生成记录，它将用户记录添加到状态存储第二个使用java生成记录到主题alert-topic，它将使用用户ID从状态存储中获取记录userIdToUserRecord.get("user-12345");

工作得很好，我正在使用 kafkaavroproducer 为这两个主题制作记录

案例 2： 首先使用 python 为主题 user-setting-topic 生成记录，它将用户记录添加到状态存储 *userIdToUserRecord.put("user-100", GenericRecord);

第二次使用java生成记录到主题alert-topic，它将使用用户ID从状态存储中获取记录userIdToUserRecord.get("user-100"); p>

这里发生的奇怪 userIdToUserRecord.get("user-100") 将返回 null

我也检查这样的场景我使用python为用户设置主题生成记录，然后触发的userSettingProcessor处理方法在调试模式下检查并尝试从状态存储userIdToUserRecord.get（“user-100”）获取用户记录它在userSettingProcessor中运行良好我能够从状态存储中获取数据

然后我使用 java 生成记录到警报主题然后尝试获取 userIdToUserRecord.get("user-100") 它将返回 null

我不知道这种奇怪的行为有人告诉我这种行为。

Python 代码：

value_schema = avro.load('user-setting.avsc')
value = {
    "user-id":"user-12345",
    "client_id":"5cfdd3db-b25a-4e21-a67d-462697096e20",
    "alert_type":"WORK_ORDER_VOLUME"
}

print("------------------------Kafka Producer------------------------------")
avroProducer = AvroProducer(
    {'bootstrap.servers': 'localhost:9092', 'schema.registry.url': 'http://localhost:8089'},
    default_value_schema=value_schema)
avroProducer.produce(topic="user-setting-topic", value=value)
print("------------------------Sucess Producer------------------------------")
avroProducer.flush()

Java 代码：

 Schema schema = new Schema.Parser().parse(schemaString);

        GenericData.Record record = new GenericData.Record(schema);
        record.put("alert_id","5cfdd3db-b25a-4e21-a67d-462697096e20");
        record.put("alert_created_at",123449437L);
        record.put("alert_type","WORK_ORDER_VOLUME");
        record.put("client_id","5cfdd3db-b25a-4e21-a67d-462697096e20");
        //record.put("property_key","property_key-"+i);

        record.put("alert_data","{\"alert_trigger_info\":{\"jll_value\":1.4,\"jll_category\":\"internal\",\"name\":\"trade_Value\",\"current_value\":40,\"calculated_value\":40.1},\"work_order\":{\"locations\":{\"country_name\":\"value\",\"state_province\":\"value\",\"city\":\"value\"},\"property\":{\"name\":\"property name\"}}}");
        return record;

【问题讨论】：

标签： java python apache-kafka-streams kafka-producer-api confluent-platform

【解决方案1】：

问题在于 Java 生产者和 Python 生产者（基于 C 生产者）使用不同的默认散列函数进行数据分区。您需要为一个（或两个）提供自定义分区，以确保它们使用相同的分区策略。

不幸的是，Kafka 协议没有指定默认的分区哈希函数应该是什么，因此客户端可以默认使用他们想要的任何东西。

【讨论】：