Kafka流-内部主题的ACL

Question

我正在尝试设置一个安全的Kafka集群，但在ACL方面遇到了一些困难。

Kafka流的汇流安全指南（https://docs.Confluent.io/current/Streams/developer-guide/html" target="_blank">security.html）只说明必须将集群创建ACL交给主体...但它没有说任何关于如何实际处理内部话题的内容。

通过研究和实验，我确定（对于Kafka版本1.0.0):

1. 通配符不能与文本一起用于ACL中的主题名称。例如，由于所有内部主题都以应用程序id为前缀，所以我首先想到的是将acl应用于匹配' -*'的主题。这不起作用。
2. 由Streams API创建的主题不会自动获得授予创建者的读/写访问权限。

内部主题的确切名称是否可预测和一致？换句话说，如果我在开发服务器上运行我的应用程序，当运行时，是否会在生产服务器上创建完全相同的主题？如果是这样，那么我可以在部署之前添加从dev派生的ACL。如果没有，应如何添加ACL？

Answer 1

内部主题的确切名称是否可预测和一致？换句话说，如果我在开发服务器上运行我的应用程序，当运行时，是否会在生产服务器上创建完全相同的主题？

是的，您将从一个运行到另一个运行获得完全相同的主题名称。DSL生成具有如下功能的处理器名称：

public String newProcessorName(final String prefix) {
    return prefix + String.format("%010d", index.getAndIncrement());
}

（其中index只是一个递增整数）。然后使用这些处理器名称创建重分区主题，其函数如下所示（参数name是如上所述生成的处理器名称）：

static <K1, V1> String createReparitionedSource(final InternalStreamsBuilder builder,
                                                final Serde<K1> keySerde,
                                                final Serde<V1> valSerde,
                                                final String topicNamePrefix,
                                                final String name) {
    Serializer<K1> keySerializer = keySerde != null ? keySerde.serializer() : null;
    Serializer<V1> valSerializer = valSerde != null ? valSerde.serializer() : null;
    Deserializer<K1> keyDeserializer = keySerde != null ? keySerde.deserializer() : null;
    Deserializer<V1> valDeserializer = valSerde != null ? valSerde.deserializer() : null;
    String baseName = topicNamePrefix != null ? topicNamePrefix : name;

    String repartitionTopic = baseName + REPARTITION_TOPIC_SUFFIX;
    String sinkName = builder.newProcessorName(SINK_NAME);
    String filterName = builder.newProcessorName(FILTER_NAME);
    String sourceName = builder.newProcessorName(SOURCE_NAME);

    builder.internalTopologyBuilder.addInternalTopic(repartitionTopic);
    builder.internalTopologyBuilder.addProcessor(filterName, new KStreamFilter<>(new Predicate<K1, V1>() {
        @Override
        public boolean test(final K1 key, final V1 value) {
            return key != null;
        }
    }, false), name);

    builder.internalTopologyBuilder.addSink(sinkName, repartitionTopic, keySerializer, valSerializer,
        null, filterName);
    builder.internalTopologyBuilder.addSource(null, sourceName, new FailOnInvalidTimestamp(),
        keyDeserializer, valDeserializer, repartitionTopic);

    return sourceName;
}

我没有使用过ACL，但我想既然这些只是常规主题，那么是的，您可以对它们应用ACL。《安全指南》确实提到：

在安全的Kafka群集上运行应用程序时，运行应用程序的主体必须设置ACL-Cluster--操作Create，以便应用程序具有创建内部主题的权限。

我自己也一直在想这个问题，所以如果我错了，我猜Confluent的人会纠正我的。