Repository: knative-sandbox/eventing-kafka
Pull request #742 (2021-06-28 17:52:17)

Lock unlock protocol flags
Subtask of #588 to add locks to ConsumerGroupAsyncCommands (dispatcher side)

Proposed Changes
🎁. Added functionality to the managedGroup (resetLockTimer/canUnlock/removeLock) to allow locking and unlocking of a managedGroup via an arbitrary token.
🎁  Changed the stopConsumerGroup and startConsumerGroup functions to call processLock, which understands the fields in the new CommandLock struct of the ConsumerGroupAsyncCommand and will lock/unlock the managedGroup
🧽  Change the Reconfigure function to also obey the managedGroup lock/unlock using an internal CommandLock token

The ResetOffset Controller is still "inactive" and has not been added to any Channels/Sources/etc.
The ResetOffset Controller is still not "complete" and this is a stepping stone in the path to implementing the feature.

Release Note
Release Notes will be provided near the end of the ResetOffset feature development.

Docs
Documentation will be provided near the end of the ResetOffset feature development.


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/742#" 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/main/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:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | 5d4369f02699c5db6d773bc273ada8eebb5defda | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/742/pull-knative-sandbox-eventing-kafka-unit-tests/1409570370087817216) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/742). [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/742?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#742](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/742?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (5d4369f) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/000dd9c4955f4b3c3e2f71b35d30138c7d1b9770?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (000dd9c) will **increase** coverage by `0.30%`.
> The diff coverage is `77.95%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/742/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/742?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@            Coverage Diff             @@
##             main     #742      +/-   ##
==========================================
+ Coverage   75.29%   75.60%   +0.30%     
==========================================
  Files         132      140       +8     
  Lines        5785     6297     +512     
==========================================
+ Hits         4356     4761     +405     
- Misses       1208     1316     +108     
+ Partials      221      220       -1     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/742?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...onsolidated/dispatcher/consumer\_message\_handler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/742/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL2Rpc3BhdGNoZXIvY29uc3VtZXJfbWVzc2FnZV9oYW5kbGVyLmdv) | `0.00% <0.00%> (ø)` | |
| [.../channel/distributed/controller/util/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/742/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29udHJvbGxlci91dGlsL2Rpc3BhdGNoZXIuZ28=) | `100.00% <ø> (ø)` | |
| [pkg/source/adapter/adapter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/742/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9hZGFwdGVyL2FkYXB0ZXIuZ28=) | `58.42% <ø> (-1.08%)` | :arrow_down: |
| [pkg/source/client/client.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/742/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9jbGllbnQvY2xpZW50Lmdv) | `15.66% <0.00%> (-19.48%)` | :arrow_down: |
| [pkg/source/reconciler/source/kafkasource.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/742/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9yZWNvbmNpbGVyL3NvdXJjZS9rYWZrYXNvdXJjZS5nbw==) | `0.00% <0.00%> (ø)` | |
| [pkg/source/mtadapter/adapter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/742/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9tdGFkYXB0ZXIvYWRhcHRlci5nbw==) | `47.82% <22.72%> (-27.36%)` | :arrow_down: |
| [pkg/apis/sources/v1beta1/kafka\_lifecycle.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/742/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2FwaXMvc291cmNlcy92MWJldGExL2thZmthX2xpZmVjeWNsZS5nbw==) | `53.33% <50.00%> (-0.52%)` | :arrow_down: |
| [pkg/channel/consolidated/utils/util.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/742/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL3V0aWxzL3V0aWwuZ28=) | `95.91% <50.00%> (-4.09%)` | :arrow_down: |
| [pkg/source/client/offsets.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/742/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9jbGllbnQvb2Zmc2V0cy5nbw==) | `60.00% <60.00%> (ø)` | |
| [pkg/channel/consolidated/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/742/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL2Rpc3BhdGNoZXIvZGlzcGF0Y2hlci5nbw==) | `61.58% <66.66%> (+0.10%)` | :arrow_up: |
| ... and [31 more](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/742/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/742?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/742?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [36d50bf...5d4369f](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/742?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:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | a1f4116e397f4be5e87435b57e95e666e744266b | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/742/pull-knative-sandbox-eventing-kafka-unit-tests/1409584506163171328) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/742). [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 -->
knative-prow-robot: @eric-sap: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | ae31fbf44c9973d85dfc436f56b06912f6d91ede | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/742/pull-knative-sandbox-eventing-kafka-unit-tests/1409594320129888256) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/742). [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 -->
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/742#" title="Author self-approved">eric-sap</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/742#pullrequestreview-694806286" title="Approved">matzew</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/main/OWNERS)~~ [eric-sap,matzew]

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":[]} -->

Patch:
diff --git a/pkg/common/consumer/consumer_manager.go b/pkg/common/consumer/consumer_manager.go
index 7510317e4..33bc40bde 100644
--- a/pkg/common/consumer/consumer_manager.go
+++ b/pkg/common/consumer/consumer_manager.go
@@ -36,6 +36,7 @@ import (
 	"context"
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/Shopify/sarama"
 	"go.uber.org/multierr"
@@ -46,6 +47,13 @@ import (
 	"knative.dev/eventing-kafka/pkg/common/controlprotocol/commands"
 )
 
+const (
+	defaultLockTimeout = 5 * time.Minute
+)
+
+// GroupLockedError is the error returned if the a locked managed group is given a different token for access
+var GroupLockedError = fmt.Errorf("managed group lock failed: locked by a different token")
+
 // KafkaConsumerGroupManager keeps track of Sarama consumer groups and handles messages from control-protocol clients
 type KafkaConsumerGroupManager interface {
 	Reconfigure(brokers []string, config *sarama.Config) error
@@ -113,16 +121,26 @@ func NewConsumerGroupManager(logger *zap.Logger, serverHandler controlprotocol.S
 func (m *kafkaConsumerGroupManagerImpl) Reconfigure(brokers []string, config *sarama.Config) error {
 	m.logger.Info("Reconfigure Consumer Group Manager - Stopping All Managed Consumer Groups")
 	var multiErr error
+	groupsToRestart := make([]string, 0, len(m.groups))
 	for groupId := range m.groups {
-		err := m.stopConsumerGroup(groupId)
+		err := m.stopConsumerGroup(getInternalLockCommand(true), groupId)
 		if err != nil {
+			// If we couldn't stop a group, or failed to obtain a lock, note it as an error.  However,
+			// in a practical sense, the new brokers/config will be used when whatever locked the group
+			// restarts it anyway.
 			multierr.AppendInto(&multiErr, err)
+		} else {
+			// Only attempt to restart groups that this function stopped
+			groupsToRestart = append(groupsToRestart, groupId)
 		}
 	}
+
 	m.factory = &kafkaConsumerGroupFactoryImpl{addrs: brokers, config: config}
+
+	// Restart any groups this function stopped
 	m.logger.Info("Reconfigure Consumer Group Manager - Starting All Managed Consumer Groups")
-	for groupId := range m.groups {
-		err := m.startConsumerGroup(groupId)
+	for _, groupId := range groupsToRestart {
+		err := m.startConsumerGroup(getInternalLockCommand(false), groupId)
 		if err != nil {
 			multierr.AppendInto(&multiErr, err)
 		}
@@ -130,6 +148,17 @@ func (m *kafkaConsumerGroupManagerImpl) Reconfigure(brokers []string, config *sa
 	return multiErr
 }
 
+// getInternalLockCommand returns a CommandLock object with a constant lock token used internally by
+// the consumer manager.
+func getInternalLockCommand(lock bool) *commands.CommandLock {
+	return &commands.CommandLock{
+		Token:       "consumer-manager-internal-token",
+		Timeout:     time.Minute,
+		LockBefore:  lock,
+		UnlockAfter: !lock,
+	}
+}
+
 // StartConsumerGroup uses the consumer factory to create a new ConsumerGroup, add it to the list
 // of managed groups (for start/stop functionality) and start the Consume loop.
 func (m *kafkaConsumerGroupManagerImpl) StartConsumerGroup(groupId string, topics []string, logger *zap.SugaredLogger, handler KafkaConsumerHandler, options ...SaramaConsumerHandlerOption) error {
@@ -142,14 +171,10 @@ func (m *kafkaConsumerGroupManagerImpl) StartConsumerGroup(groupId string, topic
 	}
 
 	ctx, cancel := context.WithCancel(context.Background())
-	// 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.
-	m.groupLock.Lock()
-	m.groups[groupId] = createManagedGroup(group, cancel)
-	m.groupLock.Unlock()
+	managedGrp := createManagedGroup(m.logger, group, cancel)
 
 	// Begin listening on the group's Errors() channel and write them to the managedGroup's errors channel
-	m.groups[groupId].transferErrors(ctx)
+	managedGrp.transferErrors(ctx)
 
 	// consume is passed in to the KafkaConsumerGroupFactory so that it will call the manager's
 	// consume() function instead of the one on the internal sarama ConsumerGroup.  This allows the
@@ -161,7 +186,11 @@ func (m *kafkaConsumerGroupManagerImpl) StartConsumerGroup(groupId string, topic
 
 	// The only thing we really want from the factory is the cancel function for the customConsumerGroup
 	customGroup := m.factory.startExistingConsumerGroup(group, consume, topics, logger, handler, options...)
-	m.groups[groupId].cancelConsume = customGroup.cancel
+	managedGrp.cancelConsume = 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.
+	m.setGroup(groupId, managedGrp)
 	return nil
 }
 
@@ -194,9 +223,7 @@ func (m *kafkaConsumerGroupManagerImpl) CloseConsumerGroup(groupId string) error
 	}
 
 	// Remove this groupId from the map so that manager functions may not be called on it
-	m.groupLock.Lock()
-	delete(m.groups, groupId)
-	m.groupLock.Unlock()
+	m.removeGroup(groupId)
 
 	return nil
 }
@@ -213,18 +240,15 @@ func (m *kafkaConsumerGroupManagerImpl) Errors(groupId string) <-chan error {
 
 // IsManaged returns true if the given groupId corresponds to a managed ConsumerGroup
 func (m *kafkaConsumerGroupManagerImpl) IsManaged(groupId string) bool {
-	if _, ok := m.groups[groupId]; ok {
-		return true
-	}
-	return false
+	return m.getGroup(groupId) != nil
 }
 
 // Consume calls the Consume method of a managed consumer group, using a loop to call it again if that
 // group is restarted by the manager.  If the Consume call is terminated by some other mechanism, the
 // result will be returned to the caller.
 func (m *kafkaConsumerGroupManagerImpl) consume(ctx context.Context, groupId string, topics []string, handler sarama.ConsumerGroupHandler) error {
-	managedGrp, ok := m.groups[groupId]
-	if !ok {
+	managedGrp := m.getGroup(groupId)
+	if managedGrp == nil {
 		return fmt.Errorf("consume called on nonexistent groupId '%s'", groupId)
 	}
 	for {
@@ -236,6 +260,7 @@ func (m *kafkaConsumerGroupManagerImpl) consume(ctx context.Context, groupId str
 			return err
 		}
 		// Wait for the managed ConsumerGroup to be restarted
+		m.logger.Debug("Consume is waiting for managed group restart")
 		if !managedGrp.waitForStart(ctx) {
 			// Context was canceled; abort
 			m.logger.Debug("Managed Consume Canceled", zap.String("GroupId", groupId))
@@ -246,64 +271,165 @@ func (m *kafkaConsumerGroupManagerImpl) consume(ctx context.Context, groupId str
 
 // stopConsumerGroups closes the managed ConsumerGroup identified by the provided groupId, and marks it
 // as "stopped" (that is, "able to be restarted" as opposed to being closed by something outside the manager)
-func (m *kafkaConsumerGroupManagerImpl) stopConsumerGroup(groupId string) error {
+func (m *kafkaConsumerGroupManagerImpl) stopConsumerGroup(lock *commands.CommandLock, groupId string) error {
 	groupLogger := m.logger.With(zap.String("GroupId", groupId))
+
+	// Lock the managedGroup before stopping it, if lock.LockBefore is true
+	if err := m.lockBefore(lock, groupId); err != nil {
+		groupLogger.Error("Failed to lock consumer group prior to stopping", zap.Error(err))
+		return err
+	}
+
 	groupLogger.Info("Stopping Managed ConsumerGroup")
-	managedGrp, ok := m.groups[groupId]
-	if ok {
-		// The managedGroup's start channel must be created (that is, the managedGroup must be marked
-		// as "stopped") before closing the internal ConsumerGroup, otherwise the consume function would
-		// return control to the factory.
-		managedGrp.createRestartChannel()
-		// Close the inner sarama ConsumerGroup, which will cause our consume() function to stop
-		// and wait for the managedGroup to start again.
-		err := managedGrp.group.Close()
-		if err != nil {
-			groupLogger.Error("Failed To Close Managed ConsumerGroup", zap.Error(err))
-			// Don't leave the start channel open if the group.Close() call failed; that would be misleading
-			managedGrp.closeRestartChannel()
-			return err
-		}
-		return nil
+
+	managedGrp := m.getGroup(groupId)
+	if managedGrp == nil {
+		groupLogger.Info("ConsumerGroup Not Managed - Ignoring Stop Request")
+		return fmt.Errorf("stop requested for consumer group not in managed list: %s", groupId)
 	}
-	groupLogger.Info("ConsumerGroup Not Managed - Ignoring Stop Request")
-	return fmt.Errorf("stop requested for consumer group not in managed list: %s", groupId)
+
+	// The managedGroup's start channel must be created (that is, the managedGroup must be marked
+	// as "stopped") before closing the internal ConsumerGroup, otherwise the consume function would
+	// return control to the factory.
+	managedGrp.createRestartChannel()
+	// Close the inner sarama ConsumerGroup, which will cause our consume() function to stop
+	// and wait for the managedGroup to start again.
+	err := managedGrp.group.Close()
+	if err != nil {
+		groupLogger.Error("Failed To Close Managed ConsumerGroup", zap.Error(err))
+		// Don't leave the start channel open if the group.Close() call failed; that would be misleading
+		managedGrp.closeRestartChannel()
+		return err
+	}
+
+	// Unlock the managedGroup after stopping it, if lock.UnlockAfter is true
+	if err = m.unlockAfter(lock, groupId); err != nil {
+		groupLogger.Error("Failed to unlock consumer group after stopping", zap.Error(err))
+		return err
+	}
+	return nil
 }
 
 // startConsumerGroups creates a new Consumer Group based on the groupId provided
-func (m *kafkaConsumerGroupManagerImpl) startConsumerGroup(groupId string) error {
+func (m *kafkaConsumerGroupManagerImpl) startConsumerGroup(lock *commands.CommandLock, groupId string) error {
 	groupLogger := m.logger.With(zap.String("GroupId", groupId))
+
+	// Lock the managedGroup before starting it, if lock.LockBefore is true
+	if err := m.lockBefore(lock, groupId); err != nil {
+		groupLogger.Error("Failed to lock consumer group prior to starting", zap.Error(err))
+		return err
+	}
+
 	groupLogger.Info("Starting Managed ConsumerGroup")
-	m.groupLock.Lock() // Don't allow m.groups to be modified while processing a start request
-	defer m.groupLock.Unlock()
-	if _, ok := m.groups[groupId]; ok {
-		group, err := m.factory.createConsumerGroup(groupId)
-		if err != nil {
-			groupLogger.Error("Failed To Restart Managed ConsumerGroup", zap.Error(err))
-			return err
+	if !m.IsManaged(groupId) {
+		groupLogger.Info("ConsumerGroup Not Managed - Ignoring Start Request")
+		return fmt.Errorf("start requested for consumer group not in managed list: %s", groupId)
+	}
+
+	group, err := m.factory.createConsumerGroup(groupId)
+	if err != nil {
+		groupLogger.Error("Failed To Restart Managed ConsumerGroup", zap.Error(err))
+		return err
+	}
+
+	managedGrp := m.getGroup(groupId)
+	managedGrp.group = group
+	managedGrp.closeRestartChannel() // Closing this allows the waitForStart function to finish
+
+	// Unlock the managedGroup after starting it, if lock.UnlockAfter is true
+	if err = m.unlockAfter(lock, groupId); err != nil {
+		groupLogger.Error("Failed to unlock consumer group after starting", zap.Error(err))
+		return err
+	}
+	return nil
+}
+
+// getGroup returns a group from the groups map using the groupLock mutex
+func (m *kafkaConsumerGroupManagerImpl) getGroup(groupId string) *managedGroup {
+	m.groupLock.RLock()
+	defer m.groupLock.RUnlock()
+	return m.groups[groupId]
+}
+
+// getGroup associates a group with a groupId in the groups map using the groupLock mutex
+func (m *kafkaConsumerGroupManagerImpl) setGroup(groupId string, group *managedGroup) {
+	m.groupLock.Lock()
+	m.groups[groupId] = group
+	m.groupLock.Unlock()
+}
+
+// getGroup removes a group from the groups map by groupId, using the groupLock mutex
+func (m *kafkaConsumerGroupManagerImpl) removeGroup(groupId string) {
+	m.groupLock.Lock()
+	delete(m.groups, groupId)
+	m.groupLock.Unlock()
+}
+
+// lockBefore will lock the managedGroup corresponding to the groupId, if lock.LockBefore is true
+func (m *kafkaConsumerGroupManagerImpl) lockBefore(lock *commands.CommandLock, groupId string) error {
+	return m.processLock(lock, groupId, true)
+}
+
+// unlockAfter will unlock the managedGroup corresponding to the groupId, if lock.UnlockAfter is true
+func (m *kafkaConsumerGroupManagerImpl) unlockAfter(lock *commands.CommandLock, groupId string) error {
+	return m.processLock(lock, groupId, false)
+}
+
+// processLock handles setting and removing the managedGroup's lock status.
+// For the managedGroup with the given groupId, if that group exists, this function will:
+// - Lock the group, if "lock" is true and cmdLock.LockBefore is true
+// - Unlock the group, if "lock" is false and cmdLock.UnlockAfter is true
+// Returns an error if the lock.Token field doesn't match an existing token in the managedGroup (for either case), as
+// this indicates that the group is locked by a different sender
+func (m *kafkaConsumerGroupManagerImpl) processLock(cmdLock *commands.CommandLock, groupId string, lock bool) error {
+	if cmdLock == nil {
+		return nil // No lock processing was requested
+	}
+
+	group := m.getGroup(groupId)
+	if group == nil {
+		return nil // Can't lock a nonexistent group
+	}
+
+	if (lock && !cmdLock.LockBefore) || (!lock && !cmdLock.UnlockAfter) {
+		return nil // If neither a lock nor unlock were requested, no need to go any further
+	}
+
+	if !group.canUnlock(cmdLock.Token) {
+		m.logger.Info("Managed group access denied; already locked with a different token",
+			zap.String("Token", cmdLock.Token), zap.String("GroupId", groupId))
+		return GroupLockedError // Already locked by a different command token
+	}
+
+	if lock && cmdLock.LockBefore {
+		timeout := cmdLock.Timeout
+		if timeout == 0 {
+			timeout = defaultLockTimeout
 		}
-		m.groups[groupId].group = group
-		m.groups[groupId].closeRestartChannel() // Closing this allows the waitForStart function to finish
-		return nil
+		group.resetLock(cmdLock.Token, timeout)
+	} else if cmdLock.UnlockAfter {
+		group.removeLock()
 	}
-	groupLogger.Info("ConsumerGroup Not Managed - Ignoring Start Request")
-	return fmt.Errorf("start requested for consumer group not in managed list: %s", groupId)
+
+	return nil // Lock succeeded
 }
 
 // processAsyncGroupNotification calls the provided groupFunction with whatever GroupId is contained
 // in the commandMessage, after verifying that the command version is correct.  It then calls the
 // appropriate Async response function on the commandMessage (NotifyFailed or NotifySuccess)
-func processAsyncGroupNotification(commandMessage ctrlservice.AsyncCommandMessage, groupFunction func(groupId string) error) {
-	if cmd, ok := commandMessage.ParsedCommand().(*commands.ConsumerGroupAsyncCommand); ok {
-		if cmd.Version != commands.ConsumerGroupAsyncCommandVersion {
-			commandMessage.NotifyFailed(fmt.Errorf("version mismatch; expected %d but got %d", commands.ConsumerGroupAsyncCommandVersion, cmd.Version))
-		} else {
-			err := groupFunction(cmd.GroupId)
-			if err != nil {
-				commandMessage.NotifyFailed(err)
-			} else {
-				commandMessage.NotifySuccess()
-			}
-		}
+func processAsyncGroupNotification(commandMessage ctrlservice.AsyncCommandMessage, groupFunction func(lock *commands.CommandLock, groupId string) error) {
+	cmd, ok := commandMessage.ParsedCommand().(*commands.ConsumerGroupAsyncCommand)
+	if !ok {
+		return
+	}
+	if cmd.Version != commands.ConsumerGroupAsyncCommandVersion {
+		commandMessage.NotifyFailed(fmt.Errorf("version mismatch; expected %d but got %d", commands.ConsumerGroupAsyncCommandVersion, cmd.Version))
+		return
+	}
+	err := groupFunction(cmd.Lock, cmd.GroupId)
+	if err != nil {
+		commandMessage.NotifyFailed(err)
+		return
 	}
+	commandMessage.NotifySuccess()
 }
diff --git a/pkg/common/consumer/managed_group.go b/pkg/common/consumer/managed_group.go
index 539444073..53a94e25d 100644
--- a/pkg/common/consumer/managed_group.go
+++ b/pkg/common/consumer/managed_group.go
@@ -19,34 +19,107 @@ package consumer
 import (
 	"context"
 	"sync/atomic"
+	"time"
 
 	"github.com/Shopify/sarama"
+	"go.uber.org/zap"
 )
 
 // managedGroup contains information about a Sarama ConsumerGroup that is required to start (i.e. re-create) it
 type managedGroup struct {
-	group              sarama.ConsumerGroup
-	errors             chan error
-	restartWaitChannel chan struct{}
-	stopped            atomic.Value // Boolean value indicating channel is stopped
-	cancelErrors       func()
-	cancelConsume      func()
+	logger             *zap.Logger
+	group              sarama.ConsumerGroup // The Sarama ConsumerGroup which is under management
+	errors             chan error           // An error channel that will replicate the errors from the Sarama ConsumerGroup
+	restartWaitChannel chan struct{}        // A channel that will be closed when a stopped group is restarted
+	stopped            atomic.Value         // Boolean value indicating that the managed group is stopped
+	cancelErrors       func()               // Called by the manager's CloseConsumerGroup to terminate the error forwarding
+	cancelConsume      func()               // Called by the manager during CloseConsumerGroup
+	lockedBy           atomic.Value         // The LockToken of the ConsumerGroupAsyncCommand that requested the lock
+	cancelLockTimeout  func()               // Called internally to stop the lock timeout when a lock is removed
 }
 
 // createManagedGroup associates a Sarama ConsumerGroup and cancel function (usually from the factory)
-// inside a new managedGroup struct
-func createManagedGroup(group sarama.ConsumerGroup, cancel func()) *managedGroup {
+// 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(logger *zap.Logger, group sarama.ConsumerGroup, cancel func()) *managedGroup {
+
 	managedGrp := managedGroup{
+		logger:       logger,
 		group:        group,
 		errors:       make(chan error),
 		cancelErrors: cancel,
 	}
-	// Initialize the atomic boolean value to false, otherwise it will be a nil interface
-	// (restartWaitChannel is only accessed when "stopped" is true)
-	managedGrp.stopped.Store(false)
+
+	// Atomic values must be initialized with their desired type before being accessed, or a nil
+	// interface will be returned that is not of that type (string or bool in these cases)
+	managedGrp.lockedBy.Store("")   // Empty token string indicates "unlocked"
+	managedGrp.stopped.Store(false) // A managed group defaults to "started" when created
+
 	return &managedGrp
 }
 
+// resetLock will set the "lockedBy" field to the given lockToken (need not be the same as the
+// existing one) and reset the timer to the provided timeout.  After the timer expires, the lockToken
+// will be set to an empty string (representing "unlocked")
+func (m *managedGroup) resetLock(lockToken string, timeout time.Duration) {
+
+	// Stop any existing timer (without releasing the lock) so that it won't inadvertently do an unlock later
+	if m.cancelLockTimeout != nil {
+		m.cancelLockTimeout()
+	}
+
+	if lockToken != "" && timeout != 0 {
+		// Use the provided lockToken, which will prevent any caller with a different token from executing commands
+		lockTimer := time.NewTimer(timeout)
+		// Create a cancel function that will be used by further calls to resetLock, or to removeLock(),
+		// for the purpose of stopping the lockTimer without clearing the token
+		ctx, cancel := context.WithCancel(context.Background())
+		m.cancelLockTimeout = cancel
+		// Mark this group as "locked" by the provided token
+		m.lockedBy.Store(lockToken)
+		m.logger.Info("Managed group locked", zap.String("token", lockToken))
+
+		// Reset the lockedBy field to an empty string when the lockTimer expires.  We create a new routine
+		// each time (instead of calling lockTimer.Reset) because an existing timer may have expired long ago
+		// and exited the goroutine.
+		go func() {
+			select {
+			case <-lockTimer.C:
+				m.logger.Debug("Managed Group lock timer expired")
+				m.removeLock()
+			case <-ctx.Done():
+				m.logger.Debug("Managed Group lock timer canceled")
+				if lockTimer.Stop() {
+					// Drain the timer channel
+					<-lockTimer.C
+				}
+			}
+		}()
+
+	} else {
+		// If a lockToken and a timeout were not both provided, remove any existing lock
+		// (locking a managed group for literally forever is not supported, although an
+		// arbitrarily long time may be used).
+		m.removeLock()
+	}
+}
+
+// canUnlock returns true if the provided token is sufficient to unlock the group (that is,
+// if it either matches the existing token, or if the existing token is empty)
+func (m *managedGroup) canUnlock(token string) bool {
+	lockToken := m.lockedBy.Load()
+	return lockToken == "" || lockToken == token
+}
+
+// removeLock sets the lockedBy token to an empty string, meaning "unlocked"
+func (m *managedGroup) removeLock() {
+	if m.lockedBy.Load() != "" {
+		m.logger.Debug("Managed Group lock removed")
+		m.lockedBy.Store("")
+		m.cancelLockTimeout() // Make sure an existing timer doesn't re-clear the token later
+	}
+}
+
 // isStopped is an accessor for the restartWaitChannel channel's status, which if non-nil indicates
 // that the KafkaConsumerGroupManager stopped ("paused") the managed ConsumerGroup (versus it having
 // been closed outside the manager)
@@ -95,6 +168,7 @@ func (m *managedGroup) waitForStart(ctx context.Context) bool {
 func (m *managedGroup) transferErrors(ctx context.Context) {
 	go func() {
 		for {
+			m.logger.Debug("Starting managed group error transfer")
 			for groupErr := range m.group.Errors() {
 				m.errors <- groupErr
 			}
@@ -106,6 +180,7 @@ func (m *managedGroup) transferErrors(ctx context.Context) {
 				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")
 			if !m.waitForStart(ctx) {
 				// Abort if the context was canceled
 				return


Test patch:
diff --git a/pkg/common/consumer/consumer_manager_test.go b/pkg/common/consumer/consumer_manager_test.go
index 6c7120612..cabbc02ac 100644
--- a/pkg/common/consumer/consumer_manager_test.go
+++ b/pkg/common/consumer/consumer_manager_test.go
@@ -23,11 +23,11 @@ import (
 	"testing"
 	"time"
 
-	"github.com/stretchr/testify/mock"
-
 	"github.com/Shopify/sarama"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
 	ctrl "knative.dev/control-protocol/pkg"
+	ctrlservice "knative.dev/control-protocol/pkg/service"
 	logtesting "knative.dev/pkg/logging/testing"
 
 	"knative.dev/eventing-kafka/pkg/common/controlprotocol/commands"
@@ -286,6 +286,8 @@ func TestNotifications(t *testing.T) {
 		expectErr   bool
 		closeErr    error
 		expectClose bool
+		lockFail    bool
+		lock        *commands.CommandLock
 	}{
 		{
 			name:    "Invalid Command Version",
@@ -340,9 +342,63 @@ func TestNotifications(t *testing.T) {
 			version:     1,
 			initialStop: true,
 		},
+		{
+			name:        "Start Group OpCode, Lock Failure",
+			opcode:      commands.StartConsumerGroupOpCode,
+			groupId:     "test-group-id",
+			version:     1,
+			lock:        &commands.CommandLock{LockBefore: true, Token: "token"},
+			initialStop: true,
+			lockFail:    true,
+			expectStop:  true,
+		},
+		{
+			name:     "Stop Group OpCode, Lock Failure",
+			opcode:   commands.StopConsumerGroupOpCode,
+			groupId:  "test-group-id",
+			version:  1,
+			lock:     &commands.CommandLock{LockBefore: true, Token: "token"},
+			lockFail: true,
+		},
+		{
+			name:        "Start Group OpCode, Unlock Failure",
+			opcode:      commands.StartConsumerGroupOpCode,
+			groupId:     "test-group-id",
+			version:     1,
+			lock:        &commands.CommandLock{UnlockAfter: true, Token: "token"},
+			initialStop: true,
+			lockFail:    true,
+		},
+		{
+			name:        "Stop Group OpCode, Unlock Failure",
+			opcode:      commands.StopConsumerGroupOpCode,
+			groupId:     "test-group-id",
+			version:     1,
+			lock:        &commands.CommandLock{UnlockAfter: true, Token: "token"},
+			expectClose: true,
+			lockFail:    true,
+			expectStop:  true,
+		},
+		{
+			name:        "Start Group OpCode, Lock Success",
+			opcode:      commands.StartConsumerGroupOpCode,
+			groupId:     "test-group-id",
+			lock:        &commands.CommandLock{LockBefore: true, Token: "token"},
+			initialStop: true,
+			version:     1,
+		},
+		{
+			name:        "Stop Group OpCode, Lock Success",
+			opcode:      commands.StopConsumerGroupOpCode,
+			groupId:     "test-group-id",
+			lock:        &commands.CommandLock{LockBefore: true, Token: "token"},
+			expectClose: true,
+			expectStop:  true,
+			version:     1,
+		},
 	} {
 		t.Run(testCase.name, func(t *testing.T) {
-			manager, group, _, serverHandler := getManagerWithMockGroup(t, testCase.groupId, testCase.factoryErr)
+			manager, group, managedGrp, serverHandler := getManagerWithMockGroup(t, testCase.groupId, testCase.factoryErr)
 			impl := manager.(*kafkaConsumerGroupManagerImpl)
 			if testCase.initialStop {
 				impl.groups[testCase.groupId].createRestartChannel()
@@ -354,11 +410,18 @@ func TestNotifications(t *testing.T) {
 			handler, ok := serverHandler.Router[testCase.opcode]
 			assert.True(t, ok)
 
+			if testCase.lockFail {
+				managedGrp.lockedBy.Store("different-lock-token")
+			} else if testCase.lock != nil {
+				managedGrp.lockedBy.Store(testCase.lock.Token)
+			}
+
 			testCommand := commands.ConsumerGroupAsyncCommand{
 				Version:   testCase.version,
 				CommandId: 1,
 				TopicName: "test-topic-name",
 				GroupId:   testCase.groupId,
+				Lock:      testCase.lock,
 			}
 			payload, err := testCommand.MarshalBinary()
 			assert.Nil(t, err)
@@ -377,6 +440,114 @@ func TestNotifications(t *testing.T) {
 	}
 }
 
+func TestProcessLock(t *testing.T) {
+	defer restoreNewConsumerGroup(newConsumerGroup) // must use if calling getManagerWithMockGroup in the test
+	const existingGroup = "existing-group"
+	const newToken = "new-token"
+	const shortTimeout = 60 * time.Millisecond
+
+	for _, testCase := range []struct {
+		name            string
+		groupId         string
+		lock            *commands.CommandLock
+		existingToken   string
+		expectLock      string
+		expectUnlock    string
+		expectErrBefore error
+		expectErrAfter  error
+		expectTimerStop bool
+	}{
+		{
+			name:    "Nil LockCommand",
+			groupId: existingGroup,
+		},
+		{
+			name:    "Nonexistent Group",
+			groupId: "nonexistent-group",
+			lock:    &commands.CommandLock{LockBefore: true, UnlockAfter: true},
+		},
+		{
+			name:            "Different Lock Token",
+			groupId:         existingGroup,
+			existingToken:   "existing-token",
+			expectLock:      "existing-token",
+			expectUnlock:    "existing-token",
+			lock:            &commands.CommandLock{LockBefore: true, UnlockAfter: true, Token: newToken},
+			expectErrBefore: GroupLockedError,
+			expectErrAfter:  GroupLockedError,
+		},
+		{
+			name:          "Same Lock Token",
+			groupId:       existingGroup,
+			existingToken: newToken,
+			expectLock:    newToken,
+			lock:          &commands.CommandLock{LockBefore: true, UnlockAfter: true, Token: newToken},
+		},
+		{
+			name:       "Zero Timeout",
+			groupId:    existingGroup,
+			lock:       &commands.CommandLock{LockBefore: true, UnlockAfter: true, Token: newToken},
+			expectLock: newToken,
+		},
+		{
+			name:            "Explicit Timeout",
+			groupId:         existingGroup,
+			lock:            &commands.CommandLock{LockBefore: true, UnlockAfter: true, Token: newToken, Timeout: shortTimeout},
+			expectLock:      newToken,
+			expectTimerStop: true,
+		},
+		{
+			name:         "LockBefore Only",
+			groupId:      existingGroup,
+			lock:         &commands.CommandLock{LockBefore: true, Token: newToken},
+			expectLock:   newToken,
+			expectUnlock: newToken,
+		},
+		{
+			name:    "UnlockAfter Only",
+			groupId: existingGroup,
+			lock:    &commands.CommandLock{UnlockAfter: true, Token: newToken},
+		},
+		{
+			name:    "No Lock Or Unlock",
+			groupId: existingGroup,
+			lock:    &commands.CommandLock{Token: newToken},
+		},
+	} {
+		t.Run(testCase.name, func(t *testing.T) {
+			time.Sleep(time.Millisecond)
+			manager, _, managedGrp, server := getManagerWithMockGroup(t, existingGroup, false)
+			if testCase.existingToken != "" {
+				managedGrp.lockedBy.Store(testCase.existingToken)
+			}
+			impl := manager.(*kafkaConsumerGroupManagerImpl)
+			errBefore := impl.processLock(testCase.lock, testCase.groupId, true)
+			assert.Equal(t, testCase.expectLock, managedGrp.lockedBy.Load())
+			if testCase.expectTimerStop {
+				time.Sleep(2 * shortTimeout)
+				assert.Equal(t, "", managedGrp.lockedBy.Load())
+			}
+			errAfter := impl.processLock(testCase.lock, testCase.groupId, false)
+			assert.Equal(t, testCase.expectUnlock, managedGrp.lockedBy.Load())
+			assert.Equal(t, testCase.expectErrBefore, errBefore)
+			assert.Equal(t, testCase.expectErrAfter, errAfter)
+
+			server.AssertExpectations(t)
+		})
+	}
+}
+
+func TestProcessAsyncGroupNotification_BadMessage(t *testing.T) {
+	cmdFunctionCalled := false
+	cmdFunction := func(lock *commands.CommandLock, groupId string) error {
+		cmdFunctionCalled = true
+		return nil
+	}
+	processAsyncGroupNotification(ctrlservice.AsyncCommandMessage{}, cmdFunction)
+	// Verify that an improper AsyncCommandMessage does not result in the cmdFunction being called
+	assert.False(t, cmdFunctionCalled)
+}
+
 // getManagerWithMockGroup creates a KafkaConsumerGroupManager and optionally seeds it with a mock consumer group
 func getManagerWithMockGroup(t *testing.T, groupId string, factoryErr bool) (KafkaConsumerGroupManager,
 	*kafkatesting.MockConsumerGroup,
@@ -392,7 +563,7 @@ func getManagerWithMockGroup(t *testing.T, groupId string, factoryErr bool) (Kaf
 	}
 	manager := NewConsumerGroupManager(logtesting.TestLogger(t).Desugar(), serverHandler, []string{}, &sarama.Config{})
 	if groupId != "" {
-		mockGrp, managedGrp := createTestGroup()
+		mockGrp, managedGrp := createTestGroup(t)
 		manager.(*kafkaConsumerGroupManagerImpl).groups[groupId] = managedGrp
 		return manager, mockGrp, managedGrp, serverHandler
 	}
@@ -410,9 +581,9 @@ func getMockServerHandler() *controltesting.MockServerHandler {
 	return server
 }
 
-func createTestGroup() (*kafkatesting.MockConsumerGroup, *managedGroup) {
+func createTestGroup(t *testing.T) (*kafkatesting.MockConsumerGroup, *managedGroup) {
 	mockGroup := kafkatesting.NewMockConsumerGroup()
-	return mockGroup, createManagedGroup(mockGroup, func() {})
+	return mockGroup, createManagedGroup(logtesting.TestLogger(t).Desugar(), mockGroup, func() {})
 }
 
 // restoreNewConsumerGroup allows a single defer call to be used for saving and restoring the newConsumerGroup wrapper
diff --git a/pkg/common/consumer/managed_group_test.go b/pkg/common/consumer/managed_group_test.go
index 03186d557..df8fbdada 100644
--- a/pkg/common/consumer/managed_group_test.go
+++ b/pkg/common/consumer/managed_group_test.go
@@ -24,6 +24,7 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
+	logtesting "knative.dev/pkg/logging/testing"
 
 	kafkatesting "knative.dev/eventing-kafka/pkg/common/kafka/testing"
 )
@@ -54,7 +55,7 @@ func TestManagedGroup(t *testing.T) {
 			// Test stop/start of a managedGroup
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()
-			group := createManagedGroup(nil, cancel)
+			group := createManagedGroup(logtesting.TestLogger(t).Desugar(), nil, cancel)
 			waitGroup := sync.WaitGroup{}
 			assert.False(t, group.isStopped())
 
@@ -81,7 +82,102 @@ func TestManagedGroup(t *testing.T) {
 
 }
 
+func TestResetLockTimer(t *testing.T) {
+	const shortTimeout = 60 * time.Millisecond
+
+	for _, testCase := range []struct {
+		name          string
+		cancelTimer   func()
+		existingToken string
+		token         string
+		timeout       time.Duration
+		existingTimer bool
+		expiredTimer  bool
+		expected      string
+	}{
+		{
+			name:          "Running Timer",
+			timeout:       shortTimeout,
+			token:         "123",
+			expected:      "123",
+			existingTimer: true,
+		},
+		{
+			name:         "Expired Timer",
+			timeout:      shortTimeout,
+			token:        "123",
+			expected:     "123",
+			expiredTimer: true,
+		},
+		{
+			name:        "Existing Cancel Function",
+			timeout:     shortTimeout,
+			token:       "123",
+			expected:    "123",
+			cancelTimer: func() {},
+		},
+		{
+			name:          "Different LockToken",
+			timeout:       shortTimeout,
+			token:         "123",
+			expected:      "123",
+			existingToken: "456",
+		},
+		{
+			name:    "Empty LockToken, With Duration",
+			timeout: shortTimeout,
+		},
+		{
+			name: "Empty LockToken, No Duration",
+		},
+		{
+			name:     "Valid LockToken, With Duration",
+			token:    "123",
+			expected: "123",
+			timeout:  shortTimeout,
+		},
+		{
+			name:     "Valid LockToken, No Duration",
+			token:    "123",
+			expected: "",
+		},
+	} {
+		t.Run(testCase.name, func(t *testing.T) {
+			time.Sleep(time.Millisecond)
+			_, managedGrp := createTestGroup(t)
+			if testCase.cancelTimer != nil {
+				managedGrp.cancelLockTimeout = testCase.cancelTimer
+			}
+			if testCase.existingTimer {
+				managedGrp.resetLock("existing-token", time.Second) // Long enough not to expire during the test
+				assert.False(t, managedGrp.canUnlock(testCase.token))
+				time.Sleep(time.Millisecond) // Let the timer loop start executing
+			} else if testCase.expiredTimer {
+				managedGrp.resetLock("existing-token", time.Microsecond) // Expire the timer immediately
+				time.Sleep(shortTimeout / 2)
+			} else {
+				assert.True(t, managedGrp.canUnlock(testCase.token))
+			}
+			if testCase.existingToken != "" {
+				managedGrp.lockedBy.Store(testCase.existingToken)
+				assert.Equal(t, testCase.existingToken == testCase.token, managedGrp.canUnlock(testCase.token))
+			}
+			managedGrp.resetLock(testCase.token, testCase.timeout)
+			if testCase.existingTimer || testCase.expiredTimer {
+				time.Sleep(time.Millisecond) // Let the timer loop finish executing
+			}
+			assert.Equal(t, testCase.expected, managedGrp.lockedBy.Load())
+			if testCase.timeout != 0 && testCase.token != "" {
+				time.Sleep(2 * testCase.timeout)
+				assert.Equal(t, "", managedGrp.lockedBy.Load())
+			}
+		})
+	}
+}
+
 func TestTransferErrors(t *testing.T) {
+	const shortTimeout = 60 * time.Millisecond
+
 	for _, testCase := range []struct {
 		name       string
 		stopGroup  bool
@@ -107,7 +203,8 @@ func TestTransferErrors(t *testing.T) {
 		},
 	} {
 		t.Run(testCase.name, func(t *testing.T) {
-			mockGrp, managedGrp := createTestGroup()
+			time.Sleep(time.Millisecond)
+			mockGrp, managedGrp := createTestGroup(t)
 			mockGrp.On("Errors").Return(mockGrp.ErrorChan)
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()
@@ -126,7 +223,7 @@ func TestTransferErrors(t *testing.T) {
 			close(mockGrp.ErrorChan)
 			mockGrp.AssertExpectations(t)
 
-			time.Sleep(5 * time.Millisecond) // Let the error handling loop move forward
+			time.Sleep(shortTimeout) // Let the error handling loop move forward
 			if testCase.startGroup {
 				// Simulate the effects of startConsumerGroup (new ConsumerGroup, same managedConsumerGroup)
 				mockGrp = kafkatesting.NewMockConsumerGroup()
@@ -134,7 +231,7 @@ func TestTransferErrors(t *testing.T) {
 				managedGrp.group = mockGrp
 				managedGrp.closeRestartChannel()
 
-				time.Sleep(5 * time.Millisecond) // Let the waitForStart function finish
+				time.Sleep(shortTimeout) // Let the waitForStart function finish
 				// Verify that errors work again after restart
 				mockGrp.ErrorChan <- fmt.Errorf("test-error-2")
 				err = <-managedGrp.errors