Repository: knative-sandbox/eventing-kafka
Pull request #1048 (2022-01-06 22:30:44)

[release-1.0] Consumer manager use factory merged errors (#1044)
* ConsumerGroup Manager-use factory merged errors channel (fix for #1043)

This is a cherry-pick of #1044

knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/1048#" title="Author self-approved">eric-sap</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka/blob/release-1.0/OWNERS)~~ [eric-sap]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
knative-prow-robot: @eric-sap: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-upgrade-tests | b267cfbb7b4dc5626f3a27fc10d0bda898d5dbfe | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/1048/pull-knative-sandbox-eventing-kafka-upgrade-tests/1479218908853440512) | true | `/test pull-knative-sandbox-eventing-kafka-upgrade-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/1048). [Your PR dashboard](https://gubernator.knative.dev/pr/eric-sap).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1048?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1048](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1048?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (1f59435) into [release-1.0](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/6d52f06c018058dc91c0dc7053d0e02875b45f79?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (6d52f06) will **increase** coverage by `0.00%`.
> The diff coverage is `84.61%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1048/graphs/tree.svg?width=650&height=150&src=pr&token=W765CT93QC&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1048?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@             Coverage Diff              @@
##           release-1.0    #1048   +/-   ##
============================================
  Coverage        75.02%   75.03%           
============================================
  Files              121      121           
  Lines             5650     5655    +5     
============================================
+ Hits              4239     4243    +4     
- Misses            1188     1189    +1     
  Partials           223      223           
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1048?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [pkg/common/consumer/consumer\_factory.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1048/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9jb25zdW1lci9jb25zdW1lcl9mYWN0b3J5Lmdv) | `77.35% <66.66%> (-0.65%)` | :arrow_down: |
| [pkg/common/consumer/consumer\_manager.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1048/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9jb25zdW1lci9jb25zdW1lcl9tYW5hZ2VyLmdv) | `97.67% <100.00%> (ø)` | |
| [pkg/common/consumer/managed\_group.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1048/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9jb25zdW1lci9tYW5hZ2VkX2dyb3VwLmdv) | `100.00% <100.00%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1048?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1048?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [6d52f06...1f59435](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/1048?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).

knative-prow-robot: @eric-sap: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-upgrade-tests | 1f5943521474b0ca474337081cec502165468e36 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/1048/pull-knative-sandbox-eventing-kafka-upgrade-tests/1479465561250336768) | true | `/test pull-knative-sandbox-eventing-kafka-upgrade-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/1048). [Your PR dashboard](https://gubernator.knative.dev/pr/eric-sap).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
eric-sap: /retest
knative-prow-robot: @eric-sap: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-upgrade-tests | 1f5943521474b0ca474337081cec502165468e36 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/1048/pull-knative-sandbox-eventing-kafka-upgrade-tests/1479498056549273600) | true | `/test pull-knative-sandbox-eventing-kafka-upgrade-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/1048). [Your PR dashboard](https://gubernator.knative.dev/pr/eric-sap).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
eric-sap: The pull-knative-sandbox-eventing-kafka-upgrade-tests looks like it's having difficulty installing v0.27.0 ?  I'm not sure if that's because 0.27 became 1.0 ?
pierDipi: @eric-sap I think we need to upgrade knative.dev/hack, `go get -u knative.dev/hack@release-1.0 && hack/update-codegen.sh`
eric-sap: I believe that update did the trick; thank you very much!


Patch:
diff --git a/go.mod b/go.mod
index f023a24ce..44570c213 100644
--- a/go.mod
+++ b/go.mod
@@ -35,7 +35,7 @@ require (
 	k8s.io/utils v0.0.0-20201110183641-67b214c5f920
 	knative.dev/control-protocol v0.0.0-20211101215039-a32569595694
 	knative.dev/eventing v0.27.0
-	knative.dev/hack v0.0.0-20211101195839-11d193bf617b
+	knative.dev/hack v0.0.0-20211122163517-fe1340f21191
 	knative.dev/pkg v0.0.0-20211101212339-96c0204a70dc
 	knative.dev/reconciler-test v0.0.0-20211101214439-9839937c9b13
 )
diff --git a/go.sum b/go.sum
index 9abfbadde..088b08292 100644
--- a/go.sum
+++ b/go.sum
@@ -1296,8 +1296,9 @@ knative.dev/control-protocol v0.0.0-20211101215039-a32569595694 h1:dcN5l0upFtQLq
 knative.dev/control-protocol v0.0.0-20211101215039-a32569595694/go.mod h1:HlIoGfErGnSVKN6NI3iG5gXxCo9EcdV8sCDGW2IObJo=
 knative.dev/eventing v0.27.0 h1:P9rqKvsCeFb8sEhfK32WFTG7awdOAa7XO6p3umSq/wU=
 knative.dev/eventing v0.27.0/go.mod h1:4ppWQEQ/2B66/YFENDmV1Gjxs4meLpz6UTUgLkkINt4=
-knative.dev/hack v0.0.0-20211101195839-11d193bf617b h1:DaW1iliZlBAwq/I8gTqVu8UnfGxyb5yR7CDsJi5jyWk=
 knative.dev/hack v0.0.0-20211101195839-11d193bf617b/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
+knative.dev/hack v0.0.0-20211122163517-fe1340f21191 h1:yRUruaFxjk6tkubtg44Nya0phUm1idKZr5bFO30AkmE=
+knative.dev/hack v0.0.0-20211122163517-fe1340f21191/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack/schema v0.0.0-20211101195839-11d193bf617b/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
 knative.dev/pkg v0.0.0-20211101212339-96c0204a70dc h1:ldjbTpoMXUTgzw0IJsAFLdyA6/6QYRvz8IGPZOknEDg=
 knative.dev/pkg v0.0.0-20211101212339-96c0204a70dc/go.mod h1:SkfDk9bWIiNZD7XtILGkG7AKVyF/M6M0bGxLgl0SYL8=
diff --git a/pkg/common/consumer/consumer_factory.go b/pkg/common/consumer/consumer_factory.go
index 4ef170cdf..9744b675f 100644
--- a/pkg/common/consumer/consumer_factory.go
+++ b/pkg/common/consumer/consumer_factory.go
@@ -102,19 +102,20 @@ func (c kafkaConsumerGroupFactoryImpl) startExistingConsumerGroup(
 	errorCh := make(chan error, 10)
 	releasedCh := make(chan bool)
 	ctx, cancel := context.WithCancel(context.Background())
+	groupLogger := logger.With(zap.Any("topics", topics), zap.String("groupId", groupID), zap.String("channel", channelRef.String()))
 
 	go func() {
 		// this is a blocking func
 		// do not proceed until the check is done
 		err := c.offsetsChecker.WaitForOffsetsInitialization(ctx, groupID, topics, logger, c.addrs, c.config)
 		if err != nil {
-			logger.Errorw("error while checking if offsets are initialized", zap.Any("topics", topics), zap.String("groupId", groupID), zap.String("channel", channelRef.String()), zap.Error(err))
+			groupLogger.Errorw("error while checking if offsets are initialized", zap.Error(err))
 			errorCh <- err
 			c.enqueue(channelRef)
 			return
 		}
 
-		logger.Debugw("all offsets are initialized", zap.Any("topics", topics), zap.Any("groupID", groupID))
+		groupLogger.Debugw("all offsets are initialized")
 
 		defer func() {
 			close(errorCh)
@@ -123,8 +124,9 @@ func (c kafkaConsumerGroupFactoryImpl) startExistingConsumerGroup(
 		for {
 			consumerHandler := NewConsumerHandler(logger, handler, errorCh, options...)
 
-			err := consume(ctx, topics, &consumerHandler)
+			err = consume(ctx, topics, &consumerHandler)
 			if err == sarama.ErrClosedConsumerGroup {
+				groupLogger.Infow("consumer group was closed", zap.Error(err))
 				return
 			}
 			if err != nil {
@@ -133,6 +135,7 @@ func (c kafkaConsumerGroupFactoryImpl) startExistingConsumerGroup(
 
 			select {
 			case <-ctx.Done():
+				groupLogger.Info("consumer group terminated gracefully")
 				return
 			default:
 			}
diff --git a/pkg/common/consumer/consumer_manager.go b/pkg/common/consumer/consumer_manager.go
index 882e4143d..293756280 100644
--- a/pkg/common/consumer/consumer_manager.go
+++ b/pkg/common/consumer/consumer_manager.go
@@ -285,9 +285,10 @@ func (m *kafkaConsumerGroupManagerImpl) StartConsumerGroup(ctx context.Context,
 		return m.consume(ctx, groupId, topics, handler)
 	}
 
-	// The only thing we really want from the factory is the cancel function for the customConsumerGroup
+	// The only things we really want from the factory are the cancel function for the customConsumerGroup
+	// and the error channel (if the error channel isn't read it will fill up and block between consume calls)
 	customGroup := m.factory.startExistingConsumerGroup(groupId, group, consume, topics, logger, handler, ref, options...)
-	managedGrp := createManagedGroup(ctx, m.logger, group, cancel, customGroup.cancel)
+	managedGrp := createManagedGroup(ctx, m.logger, group, customGroup.Errors(), cancel, customGroup.cancel)
 
 	// Add the Sarama ConsumerGroup we obtained from the factory to the managed group map,
 	// so that it can be stopped and started via control-protocol messages.
diff --git a/pkg/common/consumer/managed_group.go b/pkg/common/consumer/managed_group.go
index b71dc1a79..5cefb7045 100644
--- a/pkg/common/consumer/managed_group.go
+++ b/pkg/common/consumer/managed_group.go
@@ -63,7 +63,7 @@ type managedGroupImpl struct {
 // createManagedGroup associates a Sarama ConsumerGroup and cancel function (usually from the factory)
 // inside a new managedGroup struct.  If a timeout is given (nonzero), the lockId will be reset to an
 // empty string (i.e. "unlocked") after that time has passed.
-func createManagedGroup(ctx context.Context, logger *zap.Logger, group sarama.ConsumerGroup, cancelErrors func(), cancelConsume func()) managedGroup {
+func createManagedGroup(ctx context.Context, logger *zap.Logger, group sarama.ConsumerGroup, errors <-chan error, cancelErrors func(), cancelConsume func()) managedGroup {
 
 	managedGrp := &managedGroupImpl{
 		logger:            logger,
@@ -79,8 +79,8 @@ func createManagedGroup(ctx context.Context, logger *zap.Logger, group sarama.Co
 	managedGrp.lockedBy.Store("")   // Empty token string indicates "unlocked"
 	managedGrp.stopped.Store(false) // A managed group defaults to "started" when created
 
-	// Begin listening on the group's Errors() channel and write them to the managedGroup's errors channel
-	managedGrp.transferErrors(ctx)
+	// Begin listening on the provided errors channel and write them to the managedGroup's errors channel
+	managedGrp.transferErrors(ctx, errors)
 
 	return managedGrp
 }
@@ -142,7 +142,7 @@ func (m *managedGroupImpl) consume(ctx context.Context, topics []string, handler
 		// Call the internal sarama ConsumerGroup's Consume function directly
 		err := m.getSaramaGroup().Consume(ctx, topics, handler)
 		if !m.isStopped() {
-			m.logger.Debug("Managed Consume Finished Without Stop", zap.Error(err))
+			m.logger.Warn("Managed Consume Finished Without Stop", zap.Error(err))
 			// This ConsumerGroup wasn't stopped by the manager, so pass the error along to the caller
 			return err
 		}
@@ -313,28 +313,30 @@ func (m *managedGroupImpl) waitForStart(ctx context.Context) bool {
 	}
 }
 
-// transferErrors starts a goroutine that reads errors from the managedGroup's internal group.Errors() channel
-// and sends them to the m.errors channel.  This is done so that when the group.Errors() channel is closed during
-// a stop ("pause") of the group, the m.errors channel can remain open (so that users of the manager do not
-// receive a closed error channel during stop/start events).
-func (m *managedGroupImpl) transferErrors(ctx context.Context) {
+// transferErrors starts a goroutine that reads errors from the provided errors channel and sends them to the m.errors
+// channel.  Typically, this provided channel comes from the merged errors channel created in the KafkaConsumerGroupFactory
+// This is done so that when the group.Errors() channel is closed during a stop ("pause") of the group, the m.errors
+// channel can remain open (so that users of the manager do not receive a closed error channel during stop/start events).
+func (m *managedGroupImpl) transferErrors(ctx context.Context, errors <-chan error) {
 	go func() {
 		for {
-			m.logger.Debug("Starting managed group error transfer")
-			for groupErr := range m.getSaramaGroup().Errors() {
+			m.logger.Info("Starting managed group error transfer")
+			for groupErr := range errors {
 				m.transferredErrors <- groupErr
 			}
 			if !m.isStopped() {
 				// If the error channel was closed without the consumergroup being marked as stopped,
 				// or if we were unable to wait for the group to be restarted, that is outside
-				// of the manager's responsibility, so we are finished transferring errors.
+				// the manager's responsibility, so we are finished transferring errors.
+				m.logger.Warn("Consumer group's error channel was closed unexpectedly")
 				close(m.transferredErrors)
 				return
 			}
 			// Wait for the manager to restart the Consumer Group before calling m.group.Errors() again
-			m.logger.Debug("Error transfer is waiting for managed group restart")
+			m.logger.Info("Error transfer is waiting for managed group restart")
 			if !m.waitForStart(ctx) {
 				// Abort if the context was canceled
+				m.logger.Info("Wait for managed group restart was canceled")
 				return
 			}
 		}
diff --git a/vendor/knative.dev/hack/library.sh b/vendor/knative.dev/hack/library.sh
index 0877aace4..4af73bec5 100644
--- a/vendor/knative.dev/hack/library.sh
+++ b/vendor/knative.dev/hack/library.sh
@@ -64,6 +64,31 @@ fi
 # On a Prow job, redirect stderr to stdout so it's synchronously added to log
 (( IS_PROW )) && exec 2>&1
 
+# Return the major version of a release.
+# For example, "v0.2.1" returns "0"
+# Parameters: $1 - release version label.
+function major_version() {
+  local release="${1//v/}"
+  local tokens=(${release//\./ })
+  echo "${tokens[0]}"
+}
+
+# Return the minor version of a release.
+# For example, "v0.2.1" returns "2"
+# Parameters: $1 - release version label.
+function minor_version() {
+  local tokens=(${1//\./ })
+  echo "${tokens[1]}"
+}
+
+# Return the release build number of a release.
+# For example, "v0.2.1" returns "1".
+# Parameters: $1 - release version label.
+function patch_version() {
+  local tokens=(${1//\./ })
+  echo "${tokens[2]}"
+}
+
 # Print error message and exit 1
 # Parameters: $1..$n - error message to be displayed
 function abort() {
@@ -746,6 +771,7 @@ function current_branch() {
   # Get the branch name from Prow's env var, see https://github.com/kubernetes/test-infra/blob/master/prow/jobs.md.
   # Otherwise, try getting the current branch from git.
   (( IS_PROW )) && branch_name="${PULL_BASE_REF:-}"
+  [[ -z "${branch_name}" ]] && branch_name="${GITHUB_BASE_REF:-}"
   [[ -z "${branch_name}" ]] && branch_name="$(git rev-parse --abbrev-ref HEAD)"
   echo "${branch_name}"
 }
@@ -810,33 +836,56 @@ function shellcheck_new_files() {
   fi
 }
 
+# Note: if using Github checkout action please ensure you fetch all tags prior to calling
+# this function
+#
+# ie.
+# - uses: actions/checkout@v2
+#   with:
+#     fetch-depth: 0
+#
+# See: https://github.com/actions/checkout#fetch-all-history-for-all-tags-and-branches
 function latest_version() {
-  # This function works "best effort" and works on Prow but not necessarily locally.
-  # The problem is finding the latest release. If a release occurs on the same commit which
-  # was branched from main, then the tag will be an ancestor to any commit derived from main.
-  # That was the original logic. Additionally in a release branch, the tag is always an ancestor.
-  # However, if the release commit ends up not the first commit from main, then the tag is not
-  # an ancestor of main, so we can't use `git describe` to find the most recent versioned tag. So
-  # we just sort all the tags and find the newest versioned one.
-  # But when running locally, we cannot(?) know if the current branch is a fork of main or a fork
-  # of a release branch. That's where this function will malfunction when the last release did not
-  # occur on the first commit -- it will try to run the upgrade tests from an older version instead
-  # of the most recent release.
-  # Workarounds include:
-  # Tag the first commit of the release branch. Say release-0.75 released v0.75.0 from the second commit
-  # Then tag the first commit in common between main and release-0.75 with `v0.75`.
-  # Always name your local fork master or main.
-  if [ $(current_branch) = "master" ] || [ $(current_branch) = "main" ]; then
-    # For main branch, simply use git tag without major version, this will work even
-    # if the release tag is not in the main
-    git tag -l "v[0-9]*" | sort -r --version-sort | head -n1
+  local branch_name="$(current_branch)"
+
+  # Use the latest release for main
+  if [[ "$branch_name" == "main" ]] || [[ "$branch_name" == "master" ]]; then
+    git tag -l "*$(git tag -l "*v[0-9]*" | cut -d '-' -f2 | sort -r --version-sort | head -n1)*"
+    return
+  fi
+
+  # Ideally we shouldn't need to treat release branches differently but
+  # there are scenarios where git describe will return newer tags than
+  # the ones on the current branch
+  #
+  # ie. create a PR pulling commits from 0.24 into a release-0.23 branch
+  if [[ "$branch_name" == "release-"* ]]; then
+    # Infer major, minor version from the branch name
+    local tag="${branch_name##release-}"
   else
-    local semver=$(git describe --match "v[0-9]*" --abbrev=0)
-    local major_minor=$(echo "$semver" | cut -d. -f1-2)
+    # Nearest tag with the `knative-` prefix
+    local tag=$(git describe --abbrev=0 --match "knative-v[0-9]*")
 
-    # Get the latest patch release for the major minor
-    git tag -l "${major_minor}*" | sort -r --version-sort | head -n1
+    # Fallback to older tag scheme vX.Y.Z
+    [[ -z "${tag}" ]] && tag=$(git describe --abbrev=0 --match "v[0-9]*")
+
+    # Drop the prefix
+    tag="${tag##knative-}"
   fi
+
+  local major_version="$(major_version ${tag})"
+  local minor_version="$(minor_version ${tag})"
+
+  # Hardcode the jump back from 1.0
+  if [ "$major_version" = "1" ] && [ "$minor_version" = "0" ]; then
+    local tag_filter='v0.26*'
+  else
+    # Adjust the minor down by one
+    local tag_filter="*v$major_version.$(( minor_version - 1 ))*"
+  fi
+
+  # Get the latest patch release for the major minor
+  git tag -l "${tag_filter}" | sort -r --version-sort | head -n1
 }
 
 # Initializations that depend on previous functions.
diff --git a/vendor/knative.dev/hack/release.sh b/vendor/knative.dev/hack/release.sh
index 0f7c47324..5e11e87fa 100644
--- a/vendor/knative.dev/hack/release.sh
+++ b/vendor/knative.dev/hack/release.sh
@@ -131,22 +131,6 @@ function master_version() {
   echo "${tokens[0]}.${tokens[1]}"
 }
 
-# Return the minor version of a release.
-# For example, "v0.2.1" returns "2"
-# Parameters: $1 - release version label.
-function minor_version() {
-  local tokens=(${1//\./ })
-  echo "${tokens[1]}"
-}
-
-# Return the release build number of a release.
-# For example, "v0.2.1" returns "1".
-# Parameters: $1 - release version label.
-function patch_version() {
-  local tokens=(${1//\./ })
-  echo "${tokens[2]}"
-}
-
 # Return the short commit SHA from a release tag.
 # For example, "v20010101-deadbeef" returns "deadbeef".
 function hash_from_tag() {
@@ -622,8 +606,6 @@ function publish_to_github() {
   if [[ -n "${RELEASE_NOTES}" ]]; then
     cat "${RELEASE_NOTES}" >> "${description}"
   fi
-  git tag -a "${github_tag}" -m "${title}"
-  git_push tag "${github_tag}"
 
   # Include a tag for the go module version
   #
@@ -637,8 +619,14 @@ function publish_to_github() {
     local go_module_version="v0.$(( release_minor + 27 )).$(patch_version $TAG)"
     git tag -a "${go_module_version}" -m "${title}"
     git_push tag "${go_module_version}"
+  else
+    # Pre-1.0 - use the tag as the release tag
+    github_tag="${TAG}"
   fi
 
+  git tag -a "${github_tag}" -m "${title}"
+  git_push tag "${github_tag}"
+
   [[ -n "${RELEASE_BRANCH}" ]] && commitish="--commitish=${RELEASE_BRANCH}"
   for i in {2..0}; do
     hub_tool release create \
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 083dcdbe8..4ad584453 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1257,7 +1257,7 @@ knative.dev/eventing/test/upgrade/prober/wathola/fetcher
 knative.dev/eventing/test/upgrade/prober/wathola/forwarder
 knative.dev/eventing/test/upgrade/prober/wathola/receiver
 knative.dev/eventing/test/upgrade/prober/wathola/sender
-# knative.dev/hack v0.0.0-20211101195839-11d193bf617b
+# knative.dev/hack v0.0.0-20211122163517-fe1340f21191
 ## explicit
 knative.dev/hack
 knative.dev/hack/shell


Test patch:
diff --git a/pkg/common/consumer/consumer_manager_test.go b/pkg/common/consumer/consumer_manager_test.go
index 51b767368..38ddd4e6d 100644
--- a/pkg/common/consumer/consumer_manager_test.go
+++ b/pkg/common/consumer/consumer_manager_test.go
@@ -77,7 +77,7 @@ func TestReconfigure(t *testing.T) {
 		},
 	} {
 		t.Run(testCase.name, func(t *testing.T) {
-			manager, group, _, server := getManagerWithMockGroup(t, testCase.groupId, testCase.factoryErr)
+			manager, group, mgdGroup, server := getManagerWithMockGroup(t, testCase.groupId, testCase.factoryErr)
 			if group != nil {
 				group.On("Close").Return(testCase.closeErr)
 			}
@@ -91,6 +91,10 @@ func TestReconfigure(t *testing.T) {
 				assert.Equal(t, testCase.groupId, err.GroupIds[0])
 			}
 			server.AssertExpectations(t)
+			if mgdGroup != nil {
+				close(mgdGroup.errors())
+				time.Sleep(shortTimeout) // Allow transferErrors routine to exit
+			}
 		})
 	}
 }
@@ -199,7 +203,7 @@ func TestCloseConsumerGroup(t *testing.T) {
 		},
 	} {
 		t.Run(testCase.name, func(t *testing.T) {
-			manager, group, _, server := getManagerWithMockGroup(t, testCase.groupId, false)
+			manager, group, mgdGroup, server := getManagerWithMockGroup(t, testCase.groupId, false)
 			if group != nil {
 				group.On("Close").Return(testCase.closeErr)
 				group.On("Errors").Return(make(chan error))
@@ -207,6 +211,10 @@ func TestCloseConsumerGroup(t *testing.T) {
 			err := manager.CloseConsumerGroup(testCase.groupId)
 			assert.Equal(t, testCase.expectErr, err != nil)
 			server.AssertExpectations(t)
+			if mgdGroup != nil {
+				close(mgdGroup.errors())
+				time.Sleep(shortTimeout) // Allow transferErrors routine to exit
+			}
 		})
 	}
 }
@@ -293,7 +301,7 @@ func TestErrors(t *testing.T) {
 		},
 	} {
 		t.Run(testCase.name, func(t *testing.T) {
-			manager, _, _, server := getManagerWithMockGroup(t, testCase.groupId, false)
+			manager, _, mgdGroup, server := getManagerWithMockGroup(t, testCase.groupId, false)
 			valid := manager.IsManaged(testCase.groupId)
 			stopped := manager.IsStopped(testCase.groupId)
 			mgrErrors := manager.Errors(testCase.groupId)
@@ -301,6 +309,10 @@ func TestErrors(t *testing.T) {
 			assert.Equal(t, !testCase.expectErr, valid)
 			assert.False(t, stopped) // Not actually using a stopped group in this test
 			server.AssertExpectations(t)
+			if mgdGroup != nil {
+				close(mgdGroup.errors())
+				time.Sleep(shortTimeout) // Allow transferErrors routine to exit
+			}
 		})
 	}
 }
@@ -501,6 +513,10 @@ func TestNotifications(t *testing.T) {
 				group.AssertExpectations(t)
 			}
 			serverHandler.AssertExpectations(t)
+			if managedGrp != nil {
+				close(managedGrp.errors())
+				time.Sleep(shortTimeout) // Allow transferErrors routine to exit
+			}
 		})
 	}
 }
@@ -550,7 +566,7 @@ func TestManagerEvents(t *testing.T) {
 		},
 	} {
 		t.Run(testCase.name, func(t *testing.T) {
-			manager, _, _, _ := getManagerWithMockGroup(t, "", false)
+			manager, _, mgdGroup, _ := getManagerWithMockGroup(t, "", false)
 			impl := manager.(*kafkaConsumerGroupManagerImpl)
 
 			var notifyChannels []<-chan ManagerEvent
@@ -592,6 +608,10 @@ func TestManagerEvents(t *testing.T) {
 			manager.ClearNotifications()
 			waitGroup.Wait()
 			assert.Equal(t, testCase.expectClose, count)
+			if mgdGroup != nil {
+				close(mgdGroup.errors())
+				time.Sleep(shortTimeout) // Allow transferErrors routine to exit
+			}
 		})
 	}
 }
@@ -633,10 +653,7 @@ func getManagerWithMockGroup(t *testing.T, groupId string, factoryErr bool) (Kaf
 
 func createMockAndManagedGroups(t *testing.T) (*kafkatesting.MockConsumerGroup, *managedGroupImpl) {
 	mockGroup := kafkatesting.NewMockConsumerGroup()
-	mockGroup.On("Errors").Return(make(chan error))
-	managedGrp := createManagedGroup(context.Background(), logtesting.TestLogger(t).Desugar(), mockGroup, func() {}, func() {})
-	// let the transferErrors function start (otherwise AssertExpectations will randomly fail because Errors() isn't called)
-	time.Sleep(shortTimeout)
+	managedGrp := createManagedGroup(context.Background(), logtesting.TestLogger(t).Desugar(), mockGroup, make(chan error), func() {}, func() {})
 	return mockGroup, managedGrp.(*managedGroupImpl)
 }
 
diff --git a/pkg/common/consumer/managed_group_test.go b/pkg/common/consumer/managed_group_test.go
index 6d684065c..fd3ccb626 100644
--- a/pkg/common/consumer/managed_group_test.go
+++ b/pkg/common/consumer/managed_group_test.go
@@ -62,8 +62,9 @@ func TestManagedGroup(t *testing.T) {
 			defer cancel()
 
 			mockGroup := kafkatesting.NewMockConsumerGroup()
-			mockGroup.On("Errors").Return(make(chan error))
-			group := createManagedGroup(ctx, logtesting.TestLogger(t).Desugar(), mockGroup, cancel, func() {}).(*managedGroupImpl)
+			errorChannel := make(chan error)
+			mockGroup.On("Errors").Return(errorChannel)
+			group := createManagedGroup(ctx, logtesting.TestLogger(t).Desugar(), mockGroup, errorChannel, cancel, func() {}).(*managedGroupImpl)
 			waitGroup := sync.WaitGroup{}
 			assert.False(t, group.isStopped())
 
@@ -85,6 +86,8 @@ func TestManagedGroup(t *testing.T) {
 				cancel()
 			}
 			waitGroup.Wait() // Let the waitForStart function finish
+			close(errorChannel)
+			time.Sleep(shortTimeout) // Let the transferErrors goroutine finish
 		})
 	}
 
@@ -189,6 +192,9 @@ func TestProcessLock(t *testing.T) {
 			assert.Equal(t, testCase.expectUnlock, managedGrp.lockedBy.Load())
 			assert.Equal(t, testCase.expectErrBefore, errBefore)
 			assert.Equal(t, testCase.expectErrAfter, errAfter)
+
+			close(managedGrp.errors())
+			time.Sleep(shortTimeout) // Let the transferErrors goroutine finish
 		})
 	}
 }
@@ -280,6 +286,8 @@ func TestResetLockTimer(t *testing.T) {
 				time.Sleep(2 * testCase.timeout)
 				assert.Equal(t, "", managedGrp.lockedBy.Load())
 			}
+			close(managedGrp.errors())
+			time.Sleep(shortTimeout) // Let the transferErrors goroutine finish
 		})
 	}
 }
@@ -340,6 +348,8 @@ func TestManagedGroupConsume(t *testing.T) {
 				assert.Nil(t, mgdGroup.saramaGroup.Close()) // Stops the MockConsumerGroup's Consume() call
 			}
 			waitGroup.Wait() // Allows the goroutine with the consume call to finish
+			close(mgdGroup.errors())
+			time.Sleep(shortTimeout) // Let the transferErrors goroutine finish
 		})
 	}
 }
@@ -402,7 +412,8 @@ func TestStopStart(t *testing.T) {
 			assert.Equal(t, testCase.errStopping, err != nil)
 
 			mockGroup.AssertExpectations(t)
-
+			close(mgdGroup.errors())
+			time.Sleep(shortTimeout) // Let the transferErrors goroutine finish
 		})
 	}
 }
@@ -439,6 +450,8 @@ func TestClose(t *testing.T) {
 			assert.Nil(t, err)
 			assert.Equal(t, testCase.cancel, cancelConsumeCalled)
 			assert.Equal(t, testCase.cancel, cancelErrorsCalled)
+			close(mgdGroup.errors())
+			time.Sleep(shortTimeout) // Let the transferErrors goroutine finish
 		})
 	}
 }
@@ -471,51 +484,53 @@ func TestTransferErrors(t *testing.T) {
 		t.Run(testCase.name, func(t *testing.T) {
 			time.Sleep(time.Millisecond)
 
+			// errorChan simulates the merged error channel usually provided via customConsumerGroup.handlerErrorChannel
+			errorChan := make(chan error)
+
+			// mockGrp represents the internal Sarama ConsumerGroup
 			mockGrp := kafkatesting.NewMockConsumerGroup()
-			managedGrp := managedGroupImpl{
-				logger:            logtesting.TestLogger(t).Desugar(),
-				saramaGroup:       mockGrp,
-				transferredErrors: make(chan error),
-				groupMutex:        sync.RWMutex{},
-			}
-			managedGrp.lockedBy.Store("")
-			managedGrp.stopped.Store(false)
 
-			mockGrp.On("Errors").Return(mockGrp.ErrorChan)
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()
-			managedGrp.transferErrors(ctx)
 
-			mockGrp.ErrorChan <- fmt.Errorf("test-error")
+			// Call the transferErrors function (via createManagedGroup) with the simulated customConsumerGroup error channel
+			managedGrp := createManagedGroup(ctx, logtesting.TestLogger(t).Desugar(), mockGrp, errorChan, func() {}, func() {})
+			managedGrpImpl := managedGrp.(*managedGroupImpl)
+
+			// Send an error to the simulated customConsumerGroup error channel
+			errorChan <- fmt.Errorf("test-error")
+
+			// Verify that the error appears in the managed group's error channel
 			err := <-managedGrp.errors()
 			assert.NotNil(t, err)
 			assert.Equal(t, "test-error", err.Error())
+
 			if testCase.stopGroup {
-				managedGrp.createRestartChannel()
+				managedGrpImpl.createRestartChannel()
 			}
 			if testCase.cancel {
 				cancel()
 			}
-			close(mockGrp.ErrorChan)
 			mockGrp.AssertExpectations(t)
 
-			time.Sleep(shortTimeout) // Let the error handling loop move forward
 			if testCase.startGroup {
+				time.Sleep(shortTimeout) // Let the error handling loop move forward
 				// Simulate the effects of startConsumerGroup (new ConsumerGroup, same managedConsumerGroup)
 				mockGrp = kafkatesting.NewMockConsumerGroup()
-				mockGrp.On("Errors").Return(mockGrp.ErrorChan)
-				managedGrp.saramaGroup = mockGrp
-				managedGrp.closeRestartChannel()
+				managedGrpImpl.saramaGroup = mockGrp
+				managedGrpImpl.closeRestartChannel()
 
 				time.Sleep(shortTimeout) // Let the waitForStart function finish
-				// Verify that errors work again after restart
-				mockGrp.ErrorChan <- fmt.Errorf("test-error-2")
+
+				// Verify that error transfer continues to work after the restart
+				errorChan <- fmt.Errorf("test-error-2")
 				err = <-managedGrp.errors()
 				assert.NotNil(t, err)
 				assert.Equal(t, "test-error-2", err.Error())
-				close(mockGrp.ErrorChan)
 				mockGrp.AssertExpectations(t)
 			}
+			close(managedGrp.errors())
+			time.Sleep(shortTimeout) // Let the transferErrors goroutine finish
 		})
 	}
 }