Kafka acquisition: do not create empty events when a read error occurs (#2466)

This commit is contained in:
blotus 2023-09-13 13:20:36 +02:00 committed by GitHub
parent 0040569fa9
commit 43ef32aa8d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23

View file

@ -149,7 +149,9 @@ func (k *KafkaSource) ReadMessage(out chan types.Event) error {
return nil
}
k.logger.Errorln(fmt.Errorf("while reading %s message: %w", dataSourceName, err))
continue
}
k.logger.Tracef("got message: %s", string(m.Value))
l := types.Line{
Raw: string(m.Value),
Labels: k.Config.Labels,
@ -223,7 +225,6 @@ func (kc *KafkaConfiguration) NewTLSConfig() (*tls.Config, error) {
caCertPool.AppendCertsFromPEM(caCert)
tlsConfig.RootCAs = caCertPool
tlsConfig.BuildNameToCertificate()
return &tlsConfig, err
}