是否可以重新设置GlobalKTable的键？

Question

我想重新设置GlobalKTable的键（可能是在初始化它时，因为我相信它们只在创建之后才被读取）。

这可能吗？

我在一个Spring/JavaKafka Streams应用程序中有两个主题。第一个是不压缩的，第二个是。两者都使用Avro作为它们的键和值。

应用程序从第一个（非压缩）主题流式传输记录，并通过KStream#leftJoin从压缩主题附加附加数据。压缩的主题已作为GlobalKTable引入应用程序，通过StreamsBuilder#globalTable（）创建，并且需要保持这种方式（我需要应用程序每个实例中可用的主题所有分区的所有记录）。

我知道有人在谈论支持非主键连接(https://issues.apache.org/jira/browse/KAFKA-3705)，但据我所知，我还不能这么做。。。

@Configuration
@EnableKafkaStreams
public class StreamsConfig {

  @Autowired
  private MyCustomSerdes serdes;

  @Bean
  public KStream<AvroKeyOne, AvroValueOne> reKeyJoin(StreamsBuilder streamsBuilder) {

    GlobalKTable<AvroKeyOne, AvroValueOne> globalTable = streamsBuilder.globalTable("topicOne", Consumed.with(
      serdes.getAvroKeyOne()
      serdes.getAvroValueOne()
    ));

    KStream<AvroKeyTwo, AvroValueOne> kStream = streamsBuilder.stream("topicTwo", Consumed.with(
      serdes.getAvroKeyTwo(),
      serdes.getAvroValueOne()
    ));

    kStream.join(
      globalTable,
       /**
        * the KeyValueMapper. I need to rekey the Global table as well to the
        * corresponding String (which it's data will have) if I want this join
        * to return results
        */
      (streamKey, streamValue) -> {return streamKey.getNewStringKey()},
      (/**ValueJoiner Deal**/)
    );
  }

}

Answer 1

我想重新设置GlobalKTable的键（可能是在初始化它时，因为我相信它们只在创建之后才被读取）。

今天没有直接的支持。您已经提到了即将进行的工作，例如为非主键联接的全局表添加支持，但这还不可用。

你现在可以做的是：你可以将原来的Kafka主题重新键入（重新划分）到一个新的主题中，然后将重新键入的主题读入你的全局KTable。也许这是你的选择。