Merge branch 'main' into conversation

Makefile

@@ -100,6 +100,7 @@ verify-linter-version:
 	  echo "[!] Yours:  $(INSTALLED_LINT_VERSION)"; \
 		echo "[!] Theirs: $(GH_LINT_VERSION)"; \
 		echo "[!] Upgrade: curl -sSfL https://github.com/raw/golangci/golangci-lint/master/install.sh | sh -s -- -b $(go env GOPATH)/bin $(GH_LINT_VERSION)"; \
+                GOLANGCI_LINT=$(go env GOPATH)/bin/$(GOLANGCI_LINT)
 	  sleep 3; \
 	fi;
 

bindings/aws/kinesis/kinesis.go

@@ -23,7 +23,6 @@ import (
 	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
-	"github.com/aws/aws-sdk-go/aws/credentials"
 	"github.com/aws/aws-sdk-go/aws/request"
 	"github.com/aws/aws-sdk-go/service/kinesis"
 	"github.com/cenkalti/backoff/v4"
@@ -129,7 +128,7 @@ func (a *AWSKinesis) Init(ctx context.Context, metadata bindings.Metadata) error
 	if m.KinesisConsumerMode == SharedThroughput {
 		kclConfig := config.NewKinesisClientLibConfigWithCredential(m.ConsumerName,
 			m.StreamName, m.Region, m.ConsumerName,
-			credentials.NewStaticCredentials(m.AccessKey, m.SecretKey, ""))
+			client.Config.Credentials)
 		a.workerConfig = kclConfig
 	}
 

common/component/kafka/consumer.go

@@ -67,6 +67,15 @@ func (consumer *consumer) ConsumeClaim(session sarama.ConsumerGroupSession, clai
 	} else {
 		for {
 			select {
+			// Should return when `session.Context()` is done.
+			// If not, will raise `ErrRebalanceInProgress` or `read tcp <ip>:<port>: i/o timeout` when kafka rebalance. see:
+			// https://github.com/IBM/sarama/issues/1192
+			// Make sure the check for session context done happens before the next message is processed.
+			// There is a possibility that the pod takes some time to shutdown and in case of a poison pill message, the `retry` would get interrupted (as expected),
+			// but the next message would be processed as a result,
+			// therefore dropping the poison pill message regardless of resiliency policy.
+			case <-session.Context().Done():
+				return nil
 			case message, ok := <-claim.Messages():
 				if !ok {
 					return nil
@@ -88,11 +97,6 @@ func (consumer *consumer) ConsumeClaim(session sarama.ConsumerGroupSession, clai
 						consumer.k.logger.Errorf("Error processing Kafka message: %s/%d/%d [key=%s]. Error: %v.", message.Topic, message.Partition, message.Offset, asBase64String(message.Key), err)
 					}
 				}
-			// Should return when `session.Context()` is done.
-			// If not, will raise `ErrRebalanceInProgress` or `read tcp <ip>:<port>: i/o timeout` when kafka rebalance. see:
-			// https://github.com/IBM/sarama/issues/1192
-			case <-session.Context().Done():
-				return nil
 			}
 		}
 	}

common/component/kafka/kafka.go

@@ -14,6 +14,7 @@ limitations under the License.
 package kafka
 
 import (
+	"bytes"
 	"context"
 	"encoding/binary"
 	"errors"
@@ -263,7 +264,9 @@ func getSchemaSubject(topic string) string {
 }
 
 func (k *Kafka) DeserializeValue(message *sarama.ConsumerMessage, config SubscriptionHandlerConfig) ([]byte, error) {
-	// Null Data is valid and a tombstone record. It shouldn't be serialized
+	// Null Data is valid and a tombstone record.
+	// It shouldn't be going through schema validation and decoding
+	// Instead directly convert to JSON `null`
 	if message.Value == nil {
 		return []byte("null"), nil
 	}
@@ -354,8 +357,9 @@ func (k *Kafka) getSchemaRegistyClient() (srclient.ISchemaRegistryClient, error)
 }
 
 func (k *Kafka) SerializeValue(topic string, data []byte, metadata map[string]string) ([]byte, error) {
-	// Null Data is valid and a tombstone record. It shouldn't be serialized
-	if data == nil {
+	// Null Data is valid and a tombstone record.
+	// It should be converted to NULL and not go through schema validation & encoding
+	if bytes.Equal(data, []byte("null")) || data == nil {
 		return nil, nil
 	}
 

common/component/kafka/kafka_test.go

@@ -219,7 +219,14 @@ func TestSerializeValueCachingDisabled(t *testing.T) {
 		require.NoError(t, err)
 	})
 
-	t.Run("value published null, no error", func(t *testing.T) {
+	t.Run("value published 'null', no error", func(t *testing.T) {
+		act, err := k.SerializeValue("my-topic", []byte("null"), map[string]string{"valueSchemaType": "Avro"})
+
+		require.Nil(t, act)
+		require.NoError(t, err)
+	})
+
+	t.Run("value published nil, no error", func(t *testing.T) {
 		act, err := k.SerializeValue("my-topic", nil, map[string]string{"valueSchemaType": "Avro"})
 
 		require.Nil(t, act)