kafka kstream到globalktable的联接不能与使用的键相同

我遇到了一个非常令人沮丧的问题，试图将一个kstream（由使用kafkaproducer的java驱动程序填充）连接到一个globalktable，globalktable是从一个主题填充的，而这个主题又是使用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

这给了我一个叫做我的组织的主题，这个主题以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));

生成Organizaton类的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"}
  ]
}

注意：如上所述，使用单消息转换（smt）操作将orgid设置为主题的键。
这就是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中使用的key变量中。
下面是为这些事件设置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());
          }
        });

因此，上面的join（或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"}
  ]
}

总之：
我有一个主题的kstream，它的字符串键是orgid
我有一个主题的globalktable，它的字符串键也是orgid。
即使键在globalktable中（至少它们在globalktable下面的主题中），连接也永远不起作用
有人能帮我吗？我想弄清楚这件事。

kafka kstream到globalktable的联接不能与使用的键相同

1条答案

相关问题

热门标签

最新问答