partition

Author: VihasMakwana

libbeat/outputs/kafka/partition.go

@@ -77,7 +77,7 @@ func initPartitionStrategy(
 	}
 
 	if len(partition) > 1 {
-		return nil, false, errors.New("Too many partitioners")
+		return nil, false, errors.New("too many partitioners")
 	}
 
 	// extract partitioner from config
@@ -273,7 +273,7 @@ func makeFieldsHashPartitioner(log *logp.Logger, fields []string, dropFail bool)
 }
 
 func Hash2Partition(hash uint32, numPartitions int32) int32 {
-	p := int32(hash) & 0x7FFFFFFF
+	p := int32(hash) & 0x7FFFFFFF //nolint:gosec // Conversion from int to int32 is safe here.
 	return p % numPartitions
 }
 

x-pack/otel/extension/kafkapartitionerextension/extension.go

@@ -13,7 +13,6 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/extension"
 	"go.uber.org/zap"
-	"go.uber.org/zap/zapcore"
 
 	"github.com/elastic/elastic-agent-libs/logp"
 )
@@ -28,9 +27,11 @@ type kafkaPartitioner struct {
 }
 
 func (k *kafkaPartitioner) Start(context.Context, component.Host) error {
-	partitioner, err := makePartitioner(logp.NewLogger("", zap.WrapCore(func(zapcore.Core) zapcore.Core {
-		return k.logger.Core()
-	})), k.cfg.PartitionerConfig)
+	logger, err := logp.NewZapLogger(k.logger)
+	if err != nil {
+		return fmt.Errorf("error creating logger: %w", err)
+	}
+	partitioner, err := makePartitioner(logger, k.cfg.PartitionerConfig)
 	if err != nil {
 		return fmt.Errorf("error configuring the partitioner: %w", err)
 	}

x-pack/otel/extension/kafkapartitionerextension/partitioner.go

@@ -156,7 +156,7 @@ func makeHashKgoPartitioner() kgo.Partitioner {
 			hasher.Reset()
 			_, _ = hasher.Write(msg.Key)
 
-			return int(kafka.Hash2Partition(rand.Uint32(), int32(numPartitions)))
+			return int(kafka.Hash2Partition(rand.Uint32(), int32(numPartitions))) //nolint:gosec // Conversion from int to int32 is safe here.
 		}
 	})
 }
@@ -176,7 +176,7 @@ func makeFieldsHashPartitioner(
 			if dropFail {
 				return -1
 			}
-			return int(kafka.Hash2Partition(rand.Uint32(), int32(numPartitions)))
+			return int(kafka.Hash2Partition(rand.Uint32(), int32(numPartitions))) //nolint:gosec // Conversion from int to int32 is safe here.
 		}
 
 		hasher.Reset()
@@ -197,7 +197,7 @@ func makeFieldsHashPartitioner(
 			hash = hasher.Sum32()
 		}
 
-		return int(kafka.Hash2Partition(hash, int32(numPartitions)))
+		return int(kafka.Hash2Partition(hash, int32(numPartitions))) //nolint:gosec // Conversion from int to int32 is safe here.
 	}
 }