Merge branch 'master' into gcpsecret

Author: vangent

pubsub/kafkapubsub/kafka.go

@@ -229,20 +229,20 @@ func (t *topic) SendBatch(ctx context.Context, dms []*driver.Message) error {
 	// Convert the messages to a slice of sarama.ProducerMessage.
 	ms := make([]*sarama.ProducerMessage, 0, len(dms))
 	for _, dm := range dms {
-		var kafkaKey []byte
+		var kafkaKey sarama.Encoder
 		var headers []sarama.RecordHeader
 		for k, v := range dm.Metadata {
 			if k == t.opts.KeyName {
 				// Use this key's value as the Kafka message key instead of adding it
 				// to the headers.
-				kafkaKey = []byte(v)
+				kafkaKey = sarama.ByteEncoder(v)
 			} else {
 				headers = append(headers, sarama.RecordHeader{Key: []byte(k), Value: []byte(v)})
 			}
 		}
 		pm := &sarama.ProducerMessage{
 			Topic:   t.topicName,
-			Key:     sarama.ByteEncoder(kafkaKey),
+			Key:     kafkaKey,
 			Value:   sarama.ByteEncoder(dm.Body),
 			Headers: headers,
 		}