我需要使用信任库和密钥库与客户端建立 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 err != nil {
log.Fatal(err.Error())
}
var pemData []byte
for test, b := range blocks {
_ = test
pemData = append(pemData, pem.EncodeToMemory(b)...)
}
cert, err := tls.X509KeyPair(pemData, pemData)
if err != nil {
log.Fatal(err.Error())
}
//Truststore
caCert, err := ioutil.ReadFile("./certificates/ca.pem")
if err != nil {
log.Fatal(err)
}
caCertPool := x509.NewCertPool()
caCertPool.AppendCertsFromPEM(caCert)
config := &tls.Config{
Certificates: []tls.Certificate{cert},
RootCAs: caCertPool,
}
return config
}
呼唤远方
相关分类