Backport pr95 (#102)

kyma-incubator · Mar 31, 2020 · c2a58a0 · c2a58a0
1 parent 1585b5c
commit c2a58a0
Show file tree

Hide file tree

Showing 25 changed files with 126 additions and 58 deletions.
diff --git a/components/channel/Gopkg.lock b/components/channel/Gopkg.lock
diff --git a/components/channel/Gopkg.toml b/components/channel/Gopkg.toml
@@ -35,7 +35,7 @@
 
 [[constraint]]
    name = "github.com/kyma-incubator/knative-kafka"
-   branch = "master"
+   branch = "release-0.12"
 
 # Match Azure version in common component
 [[override]]

diff --git a/components/channel/Makefile b/components/channel/Makefile
@@ -100,4 +100,4 @@ ci-release: docker-push
 
 ci-pr: docker-build
 
-.PHONY: ci-release ci-pr
+.PHONY: ci-master ci-release ci-pr
diff --git a/...ithub.com/kyma-incubator/knative-kafka/components/common/pkg/kafka/constants/constants.go b/...ithub.com/kyma-incubator/knative-kafka/components/common/pkg/kafka/constants/constants.go
diff --git a/components/common/Makefile b/components/common/Makefile
@@ -63,4 +63,4 @@ ci-release: docker-build
 
 ci-pr: docker-build
 
-.PHONY: ci-release ci-pr
+.PHONY: ci-master ci-release ci-pr
diff --git a/components/controller/Gopkg.lock b/components/controller/Gopkg.lock
diff --git a/components/controller/Gopkg.toml b/components/controller/Gopkg.toml
@@ -140,4 +140,4 @@ required = [
 
 [[constraint]]
    name = "github.com/kyma-incubator/knative-kafka"
-   branch = "master"
+   branch = "release-0.12"
diff --git a/components/controller/Makefile b/components/controller/Makefile
@@ -117,4 +117,4 @@ ci-release: docker-push
 
 ci-pr: docker-build
 
-.PHONY: ci-release ci-pr
+.PHONY: ci-master ci-release ci-pr
diff --git a/...ithub.com/kyma-incubator/knative-kafka/components/common/pkg/kafka/constants/constants.go b/...ithub.com/kyma-incubator/knative-kafka/components/common/pkg/kafka/constants/constants.go
diff --git a/.../github.com/kyma-incubator/knative-kafka/components/common/pkg/kafka/consumer/consumer.go b/.../github.com/kyma-incubator/knative-kafka/components/common/pkg/kafka/consumer/consumer.go
diff --git a/components/dispatcher/Gopkg.lock b/components/dispatcher/Gopkg.lock
diff --git a/components/dispatcher/Gopkg.toml b/components/dispatcher/Gopkg.toml
@@ -34,7 +34,7 @@
 
 [[constraint]]
    name = "github.com/kyma-incubator/knative-kafka"
-   branch = "master"
+   branch = "release-0.12"
 
 # Match Azure version in common component
 [[override]]

diff --git a/components/dispatcher/Makefile b/components/dispatcher/Makefile
@@ -100,4 +100,4 @@ ci-release: docker-push
 
 ci-pr: docker-build
 
-.PHONY: ci-release ci-pr
+.PHONY: ci-master ci-release ci-pr
diff --git a/components/dispatcher/internal/dispatcher/dispatcher.go b/components/dispatcher/internal/dispatcher/dispatcher.go
@@ -155,7 +155,7 @@ func (d *Dispatcher) handleKafkaMessages(consumerOffset ConsumerOffset, subscrip
 				_ = d.Client.Dispatch(*cloudEvent, subscription.URI) // Ignore Errors - Dispatcher Will Retry And We're Moving On!
 
 				// Update Stored Offsets Based On The Processed Message
-				d.updateOffsets(consumerOffset.consumer, e)
+				d.updateOffsets(logger, consumerOffset.consumer, e)
 				currentTimeDuration := time.Now().Sub(consumerOffset.lastOffsetCommit)
 
 				// If "OffsetCommitCount" Number Of Messages Have Been Processed Since Last Offset Commit, Then Do One Now
@@ -186,11 +186,14 @@ func (d *Dispatcher) handleKafkaMessages(consumerOffset ConsumerOffset, subscrip
 }
 
 // Store Updated Offsets For The Partition If Consumer Still Has It Assigned
-func (d *Dispatcher) updateOffsets(consumer kafkaconsumer.ConsumerInterface, message *kafka.Message) {
+func (d *Dispatcher) updateOffsets(logger *zap.Logger, consumer kafkaconsumer.ConsumerInterface, message *kafka.Message) {
 	// Store The Updated Offsets
 	offsets := []kafka.TopicPartition{message.TopicPartition}
 	offsets[0].Offset++
-	consumer.StoreOffsets(offsets)
+	topicPartitions, err := consumer.StoreOffsets(offsets)
+	if err != nil {
+		logger.Error("Kafka Consumer Failed To Store Offsets", zap.Any("TopicPartitions", topicPartitions), zap.Error(err))
+	}
 }
 
 // Commit The Stored Offsets For Partitions Still Assigned To This Consumer

diff --git a/components/dispatcher/vendor/github.com/cloudevents/sdk-go/go.mod b/components/dispatcher/vendor/github.com/cloudevents/sdk-go/go.mod
diff --git a/components/dispatcher/vendor/github.com/cloudevents/sdk-go/go.sum b/components/dispatcher/vendor/github.com/cloudevents/sdk-go/go.sum
diff --git a/components/dispatcher/vendor/github.com/cloudevents/sdk-go/pkg/cloudevents/event_marshal.go b/components/dispatcher/vendor/github.com/cloudevents/sdk-go/pkg/cloudevents/event_marshal.go
diff --git a/components/dispatcher/vendor/github.com/cloudevents/sdk-go/pkg/cloudevents/eventcontext.go b/components/dispatcher/vendor/github.com/cloudevents/sdk-go/pkg/cloudevents/eventcontext.go
diff --git a/...onents/dispatcher/vendor/github.com/cloudevents/sdk-go/pkg/cloudevents/eventcontext_v1.go b/...onents/dispatcher/vendor/github.com/cloudevents/sdk-go/pkg/cloudevents/eventcontext_v1.go
diff --git a/components/dispatcher/vendor/github.com/cloudevents/sdk-go/pkg/cloudevents/extensions.go b/components/dispatcher/vendor/github.com/cloudevents/sdk-go/pkg/cloudevents/extensions.go
diff --git a/...spatcher/vendor/github.com/cloudevents/sdk-go/pkg/cloudevents/transport/http/transport.go b/...spatcher/vendor/github.com/cloudevents/sdk-go/pkg/cloudevents/transport/http/transport.go
Original file line number	Diff line number	Diff line change
Expand Up		@@ -100,4 +100,4 @@ ci-release: docker-push

		ci-pr: docker-build

		.PHONY: ci-release ci-pr
		.PHONY: ci-master ci-release ci-pr
Original file line number	Diff line number	Diff line change
Expand Up		@@ -63,4 +63,4 @@ ci-release: docker-build

		ci-pr: docker-build

		.PHONY: ci-release ci-pr
		.PHONY: ci-master ci-release ci-pr
Original file line number	Diff line number	Diff line change
Expand Up		@@ -117,4 +117,4 @@ ci-release: docker-push

		ci-pr: docker-build

		.PHONY: ci-release ci-pr
		.PHONY: ci-master ci-release ci-pr