Kafka KStream 到 GlobalKTable 连接不适用于使用相同的密钥

Question

我有一个非常令人沮丧的问题，我试图将一个由使用 KafkaProducer 的 java 驱动程序填充的 KStream 连接到一个从 Topic 填充的 GlobalKTable，而该 Topic 又是使用 JDBCConnector 从 MySQL 中提取数据来填充的桌子。无论我如何尝试在 KStream 和 GlobalKTable 之间进行连接，它们都以相同的值作为键，都不起作用。我的意思是 ValueJoiner 永远不会被调用。我将通过显示下面的相关配置和代码来尝试和解释。感谢您的帮助。

我正在使用最新版本的融合平台。

填充 GlobalKTable 的主题是从单个 MySQL 表中提取的：

Column Name/Type:
pk/bigint(20)
org_name/varchar(255)
orgId/varchar(10)

对此的 JDBCConnector 配置是：

name=my-demo
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
key.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://localhost:8081
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://localhost:8081
connection.url=jdbc:mysql://localhost:3306/reporting?user=root&password=XXX
table.whitelist=organisation
mode=incrementing
incrementing.column.name=pk
topic.prefix=my-
transforms=keyaddition
transforms.keyaddition.type=org.apache.kafka.connect.transforms.ValueToKey
transforms.keyaddition.fields=orgId

我正在使用命令行运行 JDBC 连接器：

connect-standalone /home/jim/platform/confluent/etc/schema-registry/connect-avro-standalone.properties /home/jim/prg/kafka/config/my.mysql.properties

这给了我一个名为 my-organisation 的主题，它以 orgId 为键......到目前为止一切都很好！ （注意，命名空间似乎不是由 JDBCConnector 设置的，但我认为这不是问题，但我不确定）

现在，代码。下面是我如何初始化和创建 GlobalKTable（显示相关代码）：

final Map<String, String> serdeConfig =
    Collections.singletonMap(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG,
        schemaRegistryUrl);

final StreamsBuilder builder = new StreamsBuilder();

final SpecificAvroSerde<Organisation> orgSerde = new SpecificAvroSerde<>();
orgSerde.configure(serdeConfig, false);

// Create the GlobalKTable from the topic that was populated using the connect-standalone command line 
final GlobalKTable<String, Organisation>
   orgs =
   builder.globalTable(ORG_TOPIC, Materialized.<String, Organisation, KeyValueStore<Bytes, byte[]>>as(ORG_STORE)
           .withKeySerde(Serdes.String())
           .withValueSerde(orgSerde));

生成 Organisaton 类的 avro 模式定义为：

{"namespace": "io.confluent.examples.streams.avro",
 "type":"record",
 "name":"Organisation",
 "fields":[
    {"name": "pk",      "type":"long"},
    {"name": "org_name",   "type":"string"},
    {"name": "orgId",   "type":"string"}
  ]
}

注意：如上所述，orgId 使用单消息转换 (SMT) 操作设置为主题的键。

所以，这就是 GlobalKTable 设置。

现在进行 KStream 设置（连接的右侧）。这与 globalKTable 具有相同的键 (orgId)。为此我使用了一个简单的驱动程序：

（用例是这个主题将包含与每个组织相关的事件）

public class UploadGenerator {

  public static void main(String[] args){
    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
          io.confluent.kafka.serializers.KafkaAvroSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
          io.confluent.kafka.serializers.KafkaAvroSerializer.class);
    props.put("schema.registry.url", "http://localhost:8081");
KafkaProducer producer = new KafkaProducer(props);

// This schema is also used in the consumer application or more specifically a class generated from it.
String mySchema = "{\"namespace\": \"io.confluent.examples.streams.avro\"," +
                      "\"type\":\"record\"," +
                      "\"name\":\"DocumentUpload\"," +
                      "\"fields\":[{\"name\":\"orgId\",\"type\":\"string\"}," +
                                  "{\"name\":\"date\",\"type\":\"long\",\"logicalType\":\"timestamp-millis\"}]}";

Schema.Parser parser = new Schema.Parser();
Schema schema = parser.parse(mySchema);

// Just using three fictional organisations with the following orgIds/keys
String[] ORG_ARRAY = {"002", "003", "004"};

long count = 0;
String key = ""; // key is the realm
while(true) {
  count++;
  try {
    TimeUnit.SECONDS.sleep(5);
  } catch (InterruptedException e) {

  }
  GenericRecord avroRecord = new GenericData.Record(schema);
  int orgId = ThreadLocalRandom.current().nextInt(0, 2 + 1);

  avroRecord.put("orgId",ORG_ARRAY[orgId]);
  avroRecord.put("date",new Date().getTime());
  key = ORG_ARRAY[orgId];

  ProducerRecord<Object, Object> record = new ProducerRecord<>("topic_uploads", key, avroRecord);
  try {
    producer.send(record);
    producer.flush();
  } catch(SerializationException e) {
    System.out.println("Exccccception was generated! + " + e.getMessage());
  } catch(Exception el) {
    System.out.println("Exception: " + el.getMessage());
  }
}
  }
}

因此，这会生成一个新事件，该事件表示由 orgId 表示的组织的上传，但也专门设置在 ProducerRecord 中使用的键变量中。

以下是为这些事件设置 KStream 的代码：

final SpecificAvroSerde<DocumentUpload> uploadSerde = new SpecificAvroSerde<>();
uploadSerde.configure(serdeConfig, false);

// Get the stream of uploads
final KStream<String, DocumentUpload> uploadStream = builder.stream(UPLOADS_TOPIC, Consumed.with(Serdes.String(), uploadSerde));

// Debug output to see the contents of the stream
uploadStream.foreach((k, v) -> System.out.println("uploadStream: Key: " + k + ", Value: " + v));

// Note, I tried to re-key the stream with the orgId field (even though it was set as the key in the driver but same problem)
final KStream<String, DocumentUpload> keyedUploadStream = uploadStream.selectKey((key, value) -> value.getOrgId());
keyedUploadStream.foreach((k, v) -> System.out.println("keyedUploadStream: Key: " + k + ", Value: " + v));

// Java 7 form used as it was easier to put in debug statements
// OrgPK is just a helper class defined in the same class 
KStream<String, OrgPk> joined = keyedUploadStream.leftJoin(orgs,
        new KeyValueMapper<String, DocumentUpload, String>() { /* derive a (potentially) new key by which to lookup against the table */
          @Override
          public String apply(String key, DocumentUpload value) {
            System.out.println("1. The key passed in is: " + key);
            System.out.println("2. The upload realm passed in is: " + value.getOrgId());
            return value.getOrgId();
          }
        },
        // THIS IS NEVER CALLED WITH A join() AND WHEN CALLED WITH A leftJoin() HAS A NULL ORGANISATION
        new ValueJoiner<DocumentUpload, Organisation, OrgPk>() {
          @Override
          public OrgPk apply(DocumentUpload leftValue, Organisation rightValue) {
            System.out.println("3. Value joiner has been called...");
            if( null == rightValue ) {
              // THIS IS ALWAYS CALLED, SO THERE IS NEVER A "MATCH"
              System.out.println("    3.1. Orgnisation is NULL");
              return new OrgPk(leftValue.getRealm(), 1L);
            }
            System.out.println("    3.1. Org is OK");
            // Never reaches here - this is the issue i.e. there is never a match
            return new OrgPk(leftValue.getOrgId(), rightValue.getPk());
          }
        });

因此，即使两个键相同，上述连接（或 leftJoin）也永远不会匹配！这是主要问题。

最后，DocumentUpload 的 avro 架构是：

{"namespace": "io.confluent.examples.streams.avro",
 "type":"record",
 "name":"DocumentUpload",
 "fields":[
    {"name": "orgId",   "type":"string"},
    {"name":"date",     "type":"long",  "logicalType":"timestamp-millis"}
  ]
}

所以，总结一下：

1. 我有一个关于主题的 KStream，其字符串键为 OrgId
2. 我有一个关于主题的 GlobalKTable，其字符串键也为 OrgId。
3. 即使键在 GlobalKTable 中（至少它们在 GlobalKTable 的主题中），连接也永远不会起作用

有人可以帮助我吗？我正在努力解决这个问题。

Answer 1

通过提供状态目录配置，我能够在 Windows/Intellij 上解决此问题 StreamsConfig.STATE_DIR_CONFIG