小编gok*_*736的帖子

我需要使用信任库和密钥库与客户端建立 TLS 连接。我的商店采用 P12 格式。我的密钥库只有一把私钥和一把公钥。我的信任库只有一个集群证书。

我的问题是，我无法连接到我的 ssl-Kafka。可能握手失败，但 kafka-go 不打印任何消息。

下面是我的代码。也许有配置问题？有谁知道如何打印错误？或者也许我的配置错误在哪里？

getKafkaReader(topic string) *kafka.Reader {

    if consumer == nil {
        consumer = kafka.NewReader(kafka.ReaderConfig{
            Brokers:     []string{kafkaConfig.Host},
            GroupID:     os.Getenv("KAFKA_CONSUMER_GROUP"),
            Topic:       topic,
            Partition:   0,
            MinBytes:    10e2, // 1KB
            MaxBytes:    10e5, // 1MB
            Dialer:      getDialer(),
        })
    }
    return consumer
}

func getDialer() *kafka.Dialer {

    dialer := &kafka.Dialer{
        Timeout:   5 * time.Second,
        DualStack: true,
        TLS:       tlsConfig(),
    }

    return dialer
}

func tlsConfig() *tls.Config {

    // Keystore
    keys, _ := ioutil.ReadFile(kafkaConfig.KeyStoreLocation)
    blocks, err := p12.ToPEM(keys, kafkaConfig.KeyStorePassword)
    if …

我有以下问题。我需要在 Kafka 中为某些选定的主题设置保留时间。我找到了一个解决方案，可以使用以下命令进行设置：

kafka-topics --zookeeper localhost:2181 --alter --topic topic-name --config retention.ms=-1

我检查了 Kafka 的 Web UI 并确认它已更改。

如果可能，我想自己在 Java 中设置保留时间，但似乎找不到合适的类/配置来设置时间。我以为我可以在 ProducerConfig 类中获取有关保留的信息，但在那里找不到。

甚至可以在 Java 中设置保留时间，如果可能，我该如何完成？

提前致谢！