消息中心上的Kafka Streams KTable配置错误

Question

现在可以在Message Hub上解决此问题

我在Kafka中创建KTable时遇到了一些麻烦。我是Kafka的新手，这可能是我问题的根源，但我仍然想在这里提问。我有一个项目，我想通过计算它们的总出现次数来跟踪不同的ID。我正在使用IBM Cloud上的Message Hub来管理我的主题，到目前为止，它的工作非常出色。

我在Message Hub上有一个主题，它产生的消息如{"ID":"123","TIMESTAMP":"1525339553", "BALANCE":"100", "AMOUNT":"4"}，到目前为止，唯一相关的键是ID。

我的Kafka代码以及Streams配置如下所示：

import org.apache.kafka.streams.StreamsConfig;

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, appId);
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, "3");    
props.put("security.protocol","SASL_SSL");
props.put("sasl.mechanism","PLAIN");
props.put("ssl.protocol","TLSv1.2");
props.put("ssl.enabled.protocols","TLSv1.2");
String saslJaasConfig = "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"USERNAME\" password=\"PASSWORD\";";
saslJaasConfig = saslJaasConfig.replace("USERNAME", user).replace("PASSWORD", password);
props.put("sasl.jaas.config",saslJaasConfig);

StreamsBuilder builder = new StreamsBuilder();

KStream<String, String> Kstreams = builder.stream(myTopic);

KTable<String, Long> eventCount = Kstreams
        .flatMapValues(value -> getID(value)) //function that retrieves the ID
        .groupBy((key, value) -> value)
        .count();

运行代码时，出现以下错误：

线程“ KTableTest-e2062d11-0b30-4ed0-82b0-00d83dcd9366-> StreamThread-1”中的异常org.apache.kafka.streams.errors.StreamsException：无法创建主题KTableTest-KSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition 。

其次是：

造成原因：java.util.concurrent.ExecutionException：org.apache.kafka.common.errors.PolicyViolationException：无效的配置：{segment.index.bytes = 52428800，segment.bytes = 52428800，cleanup.policy = delete，segment.ms = 600000}。仅允许的配置：[retention.ms，cleanup.policy]

我不知道为什么会发生此错误，以及如何解决。我构建KStream和KTable的方式是否不正确？还是bluemix上的消息中心？

解决了：

在我标记为正确答案的答案下方的评论中添加摘录。事实证明，我的StreamsConfig很好，并且（目前）在Message Hub方面存在问题，但是有一种解决方法：

事实证明，使用Kafka Streams 1.1创建分区主题时，Message Hub出现问题。在进行修复时，您需要手动创建主题KTableTest-KSTREAM-AGGREGATE-STATE-STORE-0000000003-分区。它需要与您的输入主题（myTopic）一样多的分区，并将保留时间设置为最大。修正后，我会再发表评论

非常感谢您的帮助！

Answer 1

Message Hub 对创建主题时可以使用的配置有一些限制。

从收到的PolicyViolationException中，您的Streams应用程序似乎试图使用一些我们不允许的配置：

• segment.index.bytes
• segment.bytes
• segment.ms

我猜您是在Streams配置中的某个位置设置了这些，应该将其删除。

请注意，您还需要StreamsConfig.REPLICATION_FACTOR_CONFIG在配置中将其设置为3才能使用我们文档中提到的Message Hub 。