review fix

receiver/pulsarreceiver/config.go

@@ -16,23 +16,32 @@ package pulsarreceiver
 
 import (
 	"errors"
-	"time"
 
 	"github.com/apache/pulsar-client-go/pulsar"
 	"go.opentelemetry.io/collector/config"
 )
 
 type Config struct {
 	config.ReceiverSettings `mapstructure:",squash"`
-	ServiceUrl              string `mapstructure:"service_url"`
-	Topic                   string `mapstructure:"topic"`
-	Subscription            string `mapstructure:"subscription"`
-	Encoding                string `mapstructure:"encoding"`
-	ConsumerName            string `mapstructure:"consumer_name"`
-	TLSTrustCertsFilePath   string `mapstructure:"tls_trust_certs_file_path"`
-	Insecure                bool   `mapstructure:"insecure"`
-	AuthName                string `mapstructure:"auth_name"`
-	AuthParam               string `mapstructure:"auth_param"`
+	// Configure the service URL for the Pulsar service.
+	ServiceUrl string `mapstructure:"service_url"`
+	// The topic of pulsar to consume logs,metrics,traces. (default = "otlp_traces" for traces,
+	//"otlp_metrics" for metrics, "otlp_logs" for logs)
+	Topic string `mapstructure:"topic"`
+	// The Subscription that receiver will be consuming messages from (default "otlp_subscription")
+	Subscription string `mapstructure:"subscription"`
+	// Encoding of the messages (default "otlp_proto")
+	Encoding string `mapstructure:"encoding"`
+	// Name specifies the consumer name.
+	ConsumerName string `mapstructure:"consumer_name"`
+	// Set the path to the trusted TLS certificate file
+	TLSTrustCertsFilePath string `mapstructure:"tls_trust_certs_file_path"`
+	// Configure whether the Pulsar client accept untrusted TLS certificate from broker (default: false)
+	Insecure bool `mapstructure:"insecure"`
+	//AuthName to create an authentication
+	AuthName string `mapstructure:"auth_name"`
+	//AuthParam to create an authentication
+	AuthParam string `mapstructure:"auth_param"`
 }
 
 var _ config.Receiver = (*Config)(nil)
@@ -42,18 +51,13 @@ func (cfg *Config) Validate() error {
 	return nil
 }
 
-func (cfg *Config) ClientOptions() (pulsar.ClientOptions, error) {
-	duration := 20 * time.Second
-
+func (cfg *Config) clientOptions() (pulsar.ClientOptions, error) {
 	url := cfg.ServiceUrl
 	if len(url) <= 0 {
 		url = defaultServiceUrl
 	}
 	options := pulsar.ClientOptions{
-		URL:                     url,
-		MaxConnectionsPerBroker: 2,
-		ConnectionTimeout:       duration,
-		OperationTimeout:        duration,
+		URL: url,
 	}
 
 	options.TLSAllowInsecureConnection = cfg.Insecure
@@ -72,7 +76,7 @@ func (cfg *Config) ClientOptions() (pulsar.ClientOptions, error) {
 	return options, nil
 }
 
-func (cfg *Config) ConsumerOptions() (pulsar.ConsumerOptions, error) {
+func (cfg *Config) consumerOptions() (pulsar.ConsumerOptions, error) {
 	options := pulsar.ConsumerOptions{
 		Type:             pulsar.Failover,
 		Topic:            cfg.Topic,

receiver/pulsarreceiver/factory.go

@@ -25,7 +25,9 @@ import (
 const (
 	typeStr             = "pulsar"
 	defaultEncoding     = "otlp_proto"
-	defaultTopic        = "otlp_spans"
+	defaultTraceTopic   = "otlp_spans"
+	defaultMeticsTopic  = "otlp_metrics"
+	defaultLogsTopic    = "otlp_logs"
 	defaultConsumerName = ""
 	defaultSubscription = "otlp_subscription"
 	defaultServiceUrl   = "pulsar://localhost:6650"
@@ -94,6 +96,9 @@ func (f *PulsarReceiverFactory) createTracesReceiver(
 	nextConsumer consumer.Traces,
 ) (component.TracesReceiver, error) {
 	c := cfg.(*Config)
+	if len(c.Topic) <= 0 {
+		c.Topic = defaultTraceTopic
+	}
 	r, err := newTracesReceiver(*c, set, f.tracesUnmarshalers, nextConsumer)
 	if err != nil {
 		return nil, err
@@ -108,6 +113,9 @@ func (f *PulsarReceiverFactory) createMetricsReceiver(
 	nextConsumer consumer.Metrics,
 ) (component.MetricsReceiver, error) {
 	c := cfg.(*Config)
+	if len(c.Topic) <= 0 {
+		c.Topic = defaultMeticsTopic
+	}
 	r, err := newMetricsReceiver(*c, set, f.metricsUnmarshalers, nextConsumer)
 	if err != nil {
 		return nil, err
@@ -122,6 +130,9 @@ func (f *PulsarReceiverFactory) createLogsReceiver(
 	nextConsumer consumer.Logs,
 ) (component.LogsReceiver, error) {
 	c := cfg.(*Config)
+	if len(c.Topic) <= 0 {
+		c.Topic = defaultLogsTopic
+	}
 	r, err := newLogsReceiver(*c, set, f.logsUnmarshalers, nextConsumer)
 	if err != nil {
 		return nil, err
@@ -132,7 +143,7 @@ func (f *PulsarReceiverFactory) createLogsReceiver(
 func createDefaultConfig() config.Receiver {
 	return &Config{
 		ReceiverSettings: config.NewReceiverSettings(config.NewComponentID(typeStr)),
-		Topic:            defaultTopic,
+		Topic:            "",
 		Encoding:         defaultEncoding,
 		ConsumerName:     defaultConsumerName,
 		Subscription:     defaultSubscription,

receiver/pulsarreceiver/factory_test.go

@@ -28,7 +28,7 @@ func TestCreateDefaultConfig(t *testing.T) {
 	cfg := createDefaultConfig()
 	assert.Equal(t, &Config{
 		ReceiverSettings: config.NewReceiverSettings(config.NewComponentID(typeStr)),
-		Topic:            defaultTopic,
+		Topic:            "",
 		Encoding:         defaultEncoding,
 		ConsumerName:     defaultConsumerName,
 		Subscription:     defaultSubscription,

receiver/pulsarreceiver/pulsar_receiver.go

@@ -17,6 +17,7 @@ package pulsarreceiver
 import (
 	"context"
 	"errors"
+	"time"
 
 	"github.com/apache/pulsar-client-go/pulsar"
 	"go.opentelemetry.io/collector/component"
@@ -46,7 +47,7 @@ func newTracesReceiver(config Config, set component.ReceiverCreateSettings, unma
 		return nil, errUnrecognizedEncoding
 	}
 
-	options, err := config.ClientOptions()
+	options, err := config.clientOptions()
 	if err != nil {
 		return nil, err
 	}
@@ -55,7 +56,7 @@ func newTracesReceiver(config Config, set component.ReceiverCreateSettings, unma
 		return nil, err
 	}
 
-	consumerOptions, err := config.ConsumerOptions()
+	consumerOptions, err := config.consumerOptions()
 	if err != nil {
 		return nil, err
 	}
@@ -92,26 +93,31 @@ func consumerTracesLoop(ctx context.Context, c *PulsarTracesConsumer) error {
 	traceConsumer := c.tracesConsumer
 	close(c.ready)
 
-	subscribe, err := c.client.Subscribe(c.consumerOptions)
+	_consumer, err := c.client.Subscribe(c.consumerOptions)
 	if nil != err {
 		return err
 	}
 
-	c.consumer = subscribe
+	c.consumer = _consumer
 
 	for true {
-		message, err := subscribe.Receive(ctx)
+		message, err := _consumer.Receive(ctx)
 		if err != nil {
-			return err
+			if value, ok := err.(*pulsar.Error); ok && value.Result() == pulsar.AlreadyClosedError {
+				return err
+			} else {
+				time.Sleep(time.Second)
+				continue
+			}
 		}
 
 		traces, err := unmarshaler.Unmarshal(message.Payload())
 		if err != nil {
-			return err
+			c.settings.Logger.Error("unmarshaler message failed", zap.Error(err))
 		}
 
 		if err := traceConsumer.ConsumeTraces(context.Background(), traces); err != nil {
-			return err
+			c.settings.Logger.Error("consume traces failed", zap.Error(err))
 		}
 		c.consumer.Ack(message)
 	}
@@ -145,7 +151,7 @@ func newMetricsReceiver(config Config, set component.ReceiverCreateSettings, unm
 		return nil, errUnrecognizedEncoding
 	}
 
-	options, err := config.ClientOptions()
+	options, err := config.clientOptions()
 	if err != nil {
 		return nil, err
 	}
@@ -154,7 +160,7 @@ func newMetricsReceiver(config Config, set component.ReceiverCreateSettings, unm
 		return nil, err
 	}
 
-	consumerOptions, err := config.ConsumerOptions()
+	consumerOptions, err := config.consumerOptions()
 	if err != nil {
 		return nil, err
 	}
@@ -190,24 +196,31 @@ func consumeMetricsLoop(ctx context.Context, c *PulsarMetricsConsumer) error {
 	unmarshaler := c.unmarshaler
 	close(c.ready)
 
-	subscribe, err := c.client.Subscribe(c.consumerOptions)
+	_consumer, err := c.client.Subscribe(c.consumerOptions)
 	if nil != err {
 		return err
 	}
 
-	c.consumer = subscribe
+	c.consumer = _consumer
 
 	for true {
-		message, err := subscribe.Receive(ctx)
+		message, err := _consumer.Receive(ctx)
 		if err != nil {
-			return err
+			if value, ok := err.(*pulsar.Error); ok && value.Result() == pulsar.AlreadyClosedError {
+				return err
+			} else {
+				time.Sleep(time.Second)
+				continue
+			}
 		}
+
 		metrics, err := unmarshaler.Unmarshal(message.Payload())
 		if err != nil {
-			return err
+			c.settings.Logger.Error("unmarshaler message failed", zap.Error(err))
 		}
+
 		if err := c.metricsConsumer.ConsumeMetrics(context.Background(), metrics); err != nil {
-			return err
+			c.settings.Logger.Error("consume traces failed", zap.Error(err))
 		}
 
 		c.consumer.Ack(message)
@@ -242,7 +255,7 @@ func newLogsReceiver(config Config, set component.ReceiverCreateSettings, unmars
 		return nil, errUnrecognizedEncoding
 	}
 
-	options, err := config.ClientOptions()
+	options, err := config.clientOptions()
 	if err != nil {
 		return nil, err
 	}
@@ -251,7 +264,7 @@ func newLogsReceiver(config Config, set component.ReceiverCreateSettings, unmars
 		return nil, err
 	}
 
-	consumerOptions, err := config.ConsumerOptions()
+	consumerOptions, err := config.consumerOptions()
 	if err != nil {
 		return nil, err
 	}
@@ -288,27 +301,32 @@ func consumeLogsLoop(ctx context.Context, c *PulsarLogsConsumer) error {
 	unmarshaler := c.unmarshaler
 	close(c.ready)
 
-	subscribe, err := c.client.Subscribe(c.consumerOptions)
+	_consumer, err := c.client.Subscribe(c.consumerOptions)
 
 	if nil != err {
 		return err
 	}
 
-	c.consumer = subscribe
+	c.consumer = _consumer
 
 	for true {
-		message, err := subscribe.Receive(ctx)
+		message, err := _consumer.Receive(ctx)
 		if err != nil {
-			return err
+			if value, ok := err.(*pulsar.Error); ok && value.Result() == pulsar.AlreadyClosedError {
+				return err
+			} else {
+				time.Sleep(time.Second)
+				continue
+			}
 		}
 
 		logs, err := unmarshaler.Unmarshal(message.Payload())
 		if err != nil {
-			return err
+			c.settings.Logger.Error("unmarshaler message failed", zap.Error(err))
 		}
 
 		if err := c.logsConsumer.ConsumeLogs(context.Background(), logs); err != nil {
-			return err
+			c.settings.Logger.Error("consume traces failed", zap.Error(err))
 		}
 
 		c.consumer.Ack(message)