Repository: knative-sandbox/eventing-kafka
Pull request #604 (2021-05-06 20:56:53)

[source] fix issue with offset not being initialized before session is closed
<!-- Please include the 'why' behind your changes if no issue exists -->

Make sure the KafkaSource is Ready (ie. guarantee to not lose events) before starting the consumer group. 

Without this fix, events sent to partitions with an uninitialized consumer group offset could be lost (as per def above) if these events are sent while the consumer group session is closed or is being rebalanced.

## Proposed Changes

- Make sure all consumer group offsets are initialized before starting consuming events.
- 
-

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- 🎁 Add new feature
- 🐛 Fix bug
- 🧽 Update or clean up current behavior
- 🗑️ Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
🗒️ If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note

```

**Docs**

<!--
📖 If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->
 

/cc @devguyio @slinkydeveloper 


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

This pull-request has been approved by:
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please ask for approval from **lionelvillard** after the PR has been reviewed.

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

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

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka/blob/main/OWNERS)**

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":["lionelvillard"]} -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/common/consumer/consumer_handler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/604/pull-knative-sandbox-eventing-kafka-go-coverage/1390410272484626432/artifacts/line-cov.html#file0) | 75.0% | 71.4% | -3.6
[pkg/source/adapter/adapter.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/604/pull-knative-sandbox-eventing-kafka-go-coverage/1390410272484626432/artifacts/line-cov.html#file1) | 57.4% | 40.4% | -17.0

knative-prow-robot: @lionelvillard: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-go-coverage | ad26c499be8e8fccf713425fff4c64d50eec3a8b | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/604/pull-knative-sandbox-eventing-kafka-go-coverage/1390410272484626432) | `/test pull-knative-sandbox-eventing-kafka-go-coverage`

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

<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/604?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#604](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/604?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (ad26c49) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/ee1f70166dd0421eebc14f48367f0417c401eea1?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (ee1f701) will **decrease** coverage by `0.05%`.
> The diff coverage is `14.28%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/604/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/604?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     #604      +/-   ##
==========================================
- Coverage   71.44%   71.38%   -0.06%     
==========================================
  Files         140      145       +5     
  Lines        5772     5865      +93     
==========================================
+ Hits         4124     4187      +63     
- Misses       1416     1445      +29     
- Partials      232      233       +1     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/604?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [pkg/source/adapter/adapter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/604/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9hZGFwdGVyL2FkYXB0ZXIuZ28=) | `45.94% <6.66%> (-14.55%)` | :arrow_down: |
| [pkg/common/consumer/consumer\_handler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/604/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9jb25zdW1lci9jb25zdW1lcl9oYW5kbGVyLmdv) | `69.23% <60.00%> (-2.20%)` | :arrow_down: |
| [pkg/apis/kafka/v1alpha1/register.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/604/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2FwaXMva2Fma2EvdjFhbHBoYTEvcmVnaXN0ZXIuZ28=) | `100.00% <0.00%> (ø)` | |
| [pkg/apis/kafka/v1alpha1/reset\_offset\_validation.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/604/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2FwaXMva2Fma2EvdjFhbHBoYTEvcmVzZXRfb2Zmc2V0X3ZhbGlkYXRpb24uZ28=) | `100.00% <0.00%> (ø)` | |
| [pkg/apis/kafka/v1alpha1/reset\_offset\_lifecycle.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/604/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2FwaXMva2Fma2EvdjFhbHBoYTEvcmVzZXRfb2Zmc2V0X2xpZmVjeWNsZS5nbw==) | `100.00% <0.00%> (ø)` | |
| [pkg/apis/kafka/v1alpha1/reset\_offset\_defaults.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/604/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2FwaXMva2Fma2EvdjFhbHBoYTEvcmVzZXRfb2Zmc2V0X2RlZmF1bHRzLmdv) | `100.00% <0.00%> (ø)` | |
| [pkg/apis/kafka/v1alpha1/reset\_offset\_types.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/604/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2FwaXMva2Fma2EvdjFhbHBoYTEvcmVzZXRfb2Zmc2V0X3R5cGVzLmdv) | `100.00% <0.00%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/604?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/604?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [ee1f701...ad26c49](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/604?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-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/common/consumer/consumer_handler.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/604/pull-knative-sandbox-eventing-kafka-go-coverage/1392860449321521152/artifacts/line-cov.html#file0) | 75.0% | 71.4% | -3.6
[pkg/source/adapter/adapter.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/604/pull-knative-sandbox-eventing-kafka-go-coverage/1392860449321521152/artifacts/line-cov.html#file1) | 57.4% | 67.4% | 10.0

knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/source/adapter/adapter.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/604/pull-knative-sandbox-eventing-kafka-go-coverage/1395008694164393984/artifacts/line-cov.html#file0) | 57.4% | 63.9% | 6.5

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/604#pullrequestreview-663224008" title="LGTM">slinkydeveloper</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)~~ [slinkydeveloper]

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/source/adapter/adapter.go b/pkg/source/adapter/adapter.go
index 270896fe1..858ddd25e 100644
--- a/pkg/source/adapter/adapter.go
+++ b/pkg/source/adapter/adapter.go
@@ -68,6 +68,7 @@ func NewEnvConfig() adapter.EnvConfigAccessor {
 type Adapter struct {
 	config        *AdapterConfig
 	controlServer *ctrlnetwork.ControlServer
+	saramaConfig  *sarama.Config
 
 	httpMessageSender *kncloudevents.HTTPMessageSender
 	reporter          pkgsource.StatsReporter
@@ -106,8 +107,6 @@ func (a *Adapter) Start(ctx context.Context) (err error) {
 		zap.String("Namespace", a.config.Namespace),
 	)
 
-	var options []consumer.SaramaConsumerHandlerOption
-
 	// Init control service
 	if !a.config.DisableControlServer {
 		a.controlServer, err = ctrlnetwork.StartInsecureControlServer(ctx)
@@ -115,8 +114,6 @@ func (a *Adapter) Start(ctx context.Context) (err error) {
 			return err
 		}
 		a.controlServer.MessageHandler(a)
-
-		options = append(options, consumer.WithSaramaConsumerLifecycleListener(a))
 	}
 
 	// init consumer group
@@ -124,7 +121,9 @@ func (a *Adapter) Start(ctx context.Context) (err error) {
 	if err != nil {
 		return fmt.Errorf("failed to create the config: %w", err)
 	}
+	a.saramaConfig = config
 
+	options := []consumer.SaramaConsumerHandlerOption{consumer.WithSaramaConsumerLifecycleListener(a)}
 	consumerGroupFactory := consumer.NewConsumerGroupFactory(addrs, config)
 	group, err := consumerGroupFactory.StartConsumerGroup(
 		a.config.ConsumerGroup,
@@ -216,13 +215,91 @@ func (a *Adapter) HandleServiceMessage(ctx context.Context, message ctrl.Service
 }
 
 func (a *Adapter) Setup(sess sarama.ConsumerGroupSession) {
-	if err := a.controlServer.SendAndWaitForAck(kafkasourcecontrol.NotifySetupClaimsOpCode, kafkasourcecontrol.Claims(sess.Claims())); err != nil {
-		a.logger.Warnf("Cannot send the claims update: %v", err)
+	if a.controlServer != nil {
+		if err := a.controlServer.SendAndWaitForAck(kafkasourcecontrol.NotifySetupClaimsOpCode, kafkasourcecontrol.Claims(sess.Claims())); err != nil {
+			a.logger.Warnf("Cannot send the claims update: %v", err)
+		}
+	}
+
+	// Preemptively initialize consumer group offsets to be able to mark the source as ready
+	// as soon as possible.
+	if err := a.InitOffsets(sess); err != nil {
+		a.logger.Warnf("Cannot initialized consumer group offsets: %v", err)
 	}
 }
 
 func (a *Adapter) Cleanup(sess sarama.ConsumerGroupSession) {
-	if err := a.controlServer.SendAndWaitForAck(kafkasourcecontrol.NotifyCleanupClaimsOpCode, kafkasourcecontrol.Claims(sess.Claims())); err != nil {
-		a.logger.Warnf("Cannot send the claims update: %v", err)
+	if a.controlServer != nil {
+		if err := a.controlServer.SendAndWaitForAck(kafkasourcecontrol.NotifyCleanupClaimsOpCode, kafkasourcecontrol.Claims(sess.Claims())); err != nil {
+			a.logger.Warnf("Cannot send the claims update: %v", err)
+		}
+	}
+}
+
+// InitOffsets makes sure all consumer group offsets are set.
+func (a *Adapter) InitOffsets(session sarama.ConsumerGroupSession) error {
+	if a.saramaConfig.Consumer.Offsets.Initial == sarama.OffsetNewest {
+		// We want to make sure that ALL consumer group offsets are set to avoid
+		// losing events in case the consumer group session is closed before at least one message is
+		// consumed from ALL partitions.
+		// If not, an event sent to a partition with an uninitialized offset
+		// will not be forwarded when the session is closed (or a rebalancing is in progress).
+		kafkaClient, err := sarama.NewClient(a.config.BootstrapServers, a.saramaConfig)
+		if err != nil {
+			return fmt.Errorf("failed to create a Kafka client: %w", err)
+		}
+		defer kafkaClient.Close()
+
+		kafkaAdminClient, err := sarama.NewClusterAdminFromClient(kafkaClient)
+		if err != nil {
+			return fmt.Errorf("failed to create a Kafka admin client: %w", err)
+		}
+		defer kafkaAdminClient.Close()
+
+		// Retrieve all partitions
+		topicPartitions := make(map[string][]int32)
+		for _, topic := range a.config.Topics {
+			partitions, err := kafkaClient.Partitions(topic)
+
+			if err != nil {
+				return fmt.Errorf("failed to get partitions for topic %s: %w", topic, err)
+			}
+
+			topicPartitions[topic] = partitions
+		}
+
+		// Look for uninitialized offset (-1)
+		offsets, err := kafkaAdminClient.ListConsumerGroupOffsets(a.config.ConsumerGroup, topicPartitions)
+		if err != nil {
+			return err
+		}
+
+		dirty := false
+		for topic, partitions := range offsets.Blocks {
+			for partition, block := range partitions {
+				if block.Offset == -1 { // not initialized?
+
+					// Fetch the newest offset in the topic/partition and set it in the consumer group
+					offset, err := kafkaClient.GetOffset(topic, partition, sarama.OffsetNewest)
+					if err != nil {
+						return fmt.Errorf("failed to get the offset for topic %s and partition %d: %w", topic, partition, err)
+					}
+
+					a.logger.Infow("initializing offset", zap.String("topic", topic), zap.Int32("partition", partition), zap.Int64("offset", offset))
+
+					session.MarkOffset(topic, partition, offset, "")
+					dirty = true
+				}
+			}
+		}
+
+		if dirty {
+			session.Commit()
+
+			a.logger.Infow("consumer group offsets committed", zap.String("consumergroup", a.config.ConsumerGroup))
+		}
 	}
+
+	// At this stage the KafkaSource instance is considered Ready (TODO: update KafkaSource status)
+	return nil
 }


Test patch:
diff --git a/pkg/source/adapter/adapter_test.go b/pkg/source/adapter/adapter_test.go
index fdaf3843a..4c3dd8a4e 100644
--- a/pkg/source/adapter/adapter_test.go
+++ b/pkg/source/adapter/adapter_test.go
@@ -442,3 +442,196 @@ func TestAdapter_Start(t *testing.T) { // just increase code coverage
 	}
 	cancel()
 }
+
+func TestInitOffset(t *testing.T) {
+	testCases := map[string]struct {
+		topics       []string
+		topicOffsets map[string]map[int32]int64
+		cgOffsets    map[string]map[int32]int64
+		wantCommit   bool
+	}{
+		"one topic, one partition, initialized": {
+			topics: []string{"my-topic"},
+			topicOffsets: map[string]map[int32]int64{
+				"my-topic": {
+					0: 5,
+				},
+			},
+			cgOffsets: map[string]map[int32]int64{
+				"my-topic": {
+					0: 2,
+				},
+			},
+			wantCommit: false,
+		},
+		"one topic, one partition, uninitialized": {
+			topics: []string{"my-topic"},
+			topicOffsets: map[string]map[int32]int64{
+				"my-topic": {
+					0: 5,
+				},
+			},
+			cgOffsets: map[string]map[int32]int64{
+				"my-topic": {
+					0: -1,
+				},
+			},
+			wantCommit: true,
+		},
+		"several topics, several partitions, not all initialized": {
+			topics: []string{"my-topic", "my-topic-2", "my-topic-3"},
+			topicOffsets: map[string]map[int32]int64{
+				"my-topic":   {0: 5, 1: 7},
+				"my-topic-2": {0: 5, 1: 7, 2: 9},
+				"my-topic-3": {0: 5, 1: 7, 2: 2, 3: 10},
+			},
+			cgOffsets: map[string]map[int32]int64{
+				"my-topic":   {0: -1, 1: 7},
+				"my-topic-2": {0: 5, 1: -1, 2: -1},
+				"my-topic-3": {0: 5, 1: 7, 2: -1, 3: 10},
+			},
+			wantCommit: true,
+		},
+	}
+
+	for n, tc := range testCases {
+		t.Run(n, func(t *testing.T) {
+
+			broker := sarama.NewMockBroker(t, 1)
+			defer broker.Close()
+
+			group := "my-group"
+
+			offsetResponse := sarama.NewMockOffsetResponse(t).SetVersion(1)
+			for topic, partitions := range tc.topicOffsets {
+				for partition, offset := range partitions {
+					offsetResponse = offsetResponse.SetOffset(topic, partition, -1, offset)
+				}
+			}
+
+			offsetFetchResponse := sarama.NewMockOffsetFetchResponse(t).SetError(sarama.ErrNoError)
+			for topic, partitions := range tc.cgOffsets {
+				for partition, offset := range partitions {
+					offsetFetchResponse = offsetFetchResponse.SetOffset(group, topic, partition, offset, "", sarama.ErrNoError)
+				}
+			}
+
+			metadataResponse := sarama.NewMockMetadataResponse(t).
+				SetController(broker.BrokerID()).
+				SetBroker(broker.Addr(), broker.BrokerID())
+			for topic, partitions := range tc.topicOffsets {
+				for partition, _ := range partitions {
+					metadataResponse = metadataResponse.SetLeader(topic, partition, broker.BrokerID())
+				}
+			}
+
+			broker.SetHandlerByMap(map[string]sarama.MockResponse{
+				"OffsetRequest":      offsetResponse,
+				"OffsetFetchRequest": offsetFetchResponse,
+
+				"FindCoordinatorRequest": sarama.NewMockFindCoordinatorResponse(t).
+					SetCoordinator(sarama.CoordinatorGroup, group, broker),
+
+				"MetadataRequest": metadataResponse,
+			})
+
+			adapter := NewAdapter(context.Background(), NewEnvConfig(), nil, nil).(*Adapter)
+			adapter.config.ConsumerGroup = group
+			adapter.config.Topics = tc.topics
+			adapter.config.BootstrapServers = []string{broker.Addr()}
+
+			config := sarama.NewConfig()
+			config.Version = sarama.MaxVersion
+			adapter.saramaConfig = config
+
+			session := new(sampleConsumerSession)
+			err := adapter.InitOffsets(session)
+			if err != nil {
+				t.Errorf("unexpected error: %v", err)
+			}
+
+			// Check uninitialized offsets have been processed.
+			for topic, partitions := range tc.cgOffsets {
+				for partition, offset := range partitions {
+					if offset == -1 {
+						got := session.GetOffset(topic, partition)
+
+						if got != tc.topicOffsets[topic][partition] {
+							t.Errorf("wanted partition offset %d, got %d", tc.topicOffsets[topic][partition], got)
+						}
+
+					}
+				}
+			}
+
+			if tc.wantCommit && !session.committed {
+				t.Errorf("wanted offsets to be committed")
+			}
+
+			if !tc.wantCommit && session.committed {
+				t.Errorf("wanted offsets to not be committed")
+			}
+		})
+	}
+}
+
+type sampleConsumerSession struct {
+	offsets   map[string]map[int32]int64
+	committed bool
+}
+
+func (s *sampleConsumerSession) Claims() map[string][]int32 {
+	return nil
+}
+
+func (s *sampleConsumerSession) MemberID() string {
+	return ""
+}
+
+func (s *sampleConsumerSession) GenerationID() int32 {
+	return int32(0)
+}
+
+func (s *sampleConsumerSession) GetOffset(topic string, partition int32) int64 {
+	if s.offsets == nil {
+		return -2
+	}
+	partitions, ok := s.offsets[topic]
+	if !ok {
+		return -2
+	}
+	offset, ok := partitions[partition]
+	if !ok {
+		return -2
+	}
+	return offset
+}
+
+func (s *sampleConsumerSession) MarkOffset(topic string, partition int32, offset int64, metadata string) {
+	if s.offsets == nil {
+		s.offsets = make(map[string]map[int32]int64)
+	}
+
+	partitions, ok := s.offsets[topic]
+	if !ok {
+		partitions = make(map[int32]int64)
+		s.offsets[topic] = partitions
+	}
+	partitions[partition] = offset
+}
+
+func (s *sampleConsumerSession) Commit() {
+	s.committed = true
+}
+
+func (s *sampleConsumerSession) ResetOffset(topic string, partition int32, offset int64, metadata string) {
+
+}
+
+func (s *sampleConsumerSession) MarkMessage(msg *sarama.ConsumerMessage, metadata string) {
+
+}
+
+func (s *sampleConsumerSession) Context() context.Context {
+	return nil
+}