pubsublite/internal/wire/acks_test.go - gocloud - Git at Google

 // Copyright 2020 Google LLC
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
 //     https://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and

 package wire

 import (
 	"testing"

 	"cloud.google.com/go/pubsublite/internal/test"
 )

 func emptyAckConsumer(_ *ackConsumer) {
 	// Nothing to do.
 }

 func TestAckConsumerAck(t *testing.T) {
 	numAcks := 0
 	onAck := func(ac *ackConsumer) {
 		numAcks++
 	}
 	ackConsumer := newAckConsumer(0, 0, onAck)
 	if got, want := ackConsumer.IsAcked(), false; got != want {
 		t.Errorf("ackConsumer.IsAcked() got %v, want %v", got, want)
 	}

 	// Test duplicate acks.
 	for i := 0; i < 3; i++ {
 		ackConsumer.Ack()

 		if got, want := ackConsumer.IsAcked(), true; got != want {
 			t.Errorf("ackConsumer.IsAcked() got %v, want %v", got, want)
 		}
 		if got, want := numAcks, 1; got != want {
 			t.Errorf("onAck func called %v times, expected %v call", got, want)
 		}
 	}
 }

 func TestAckConsumerClear(t *testing.T) {
 	onAck := func(ac *ackConsumer) {
 		t.Error("onAck func should not have been called")
 	}
 	ackConsumer := newAckConsumer(0, 0, onAck)
 	ackConsumer.Clear()
 	ackConsumer.Ack()

 	if got, want := ackConsumer.IsAcked(), true; got != want {
 		t.Errorf("ackConsumer.IsAcked() got %v, want %v", got, want)
 	}
 }

 func TestAckTrackerProcessing(t *testing.T) {
 	ackTracker := newAckTracker()

 	// No messages received yet.
 	if got, want := ackTracker.CommitOffset(), nilCursorOffset; got != want {
 		t.Errorf("ackTracker.CommitOffset() got %v, want %v", got, want)
 	}

 	ack1 := newAckConsumer(1, 0, emptyAckConsumer)
 	ack2 := newAckConsumer(2, 0, emptyAckConsumer)
 	ack3 := newAckConsumer(3, 0, emptyAckConsumer)
 	if err := ackTracker.Push(ack1); err != nil {
 		t.Errorf("ackTracker.Push() got err %v", err)
 	}
 	if err := ackTracker.Push(ack2); err != nil {
 		t.Errorf("ackTracker.Push() got err %v", err)
 	}
 	if err := ackTracker.Push(ack3); err != nil {
 		t.Errorf("ackTracker.Push() got err %v", err)
 	}

 	// All messages unacked.
 	if got, want := ackTracker.CommitOffset(), nilCursorOffset; got != want {
 		t.Errorf("ackTracker.CommitOffset() got %v, want %v", got, want)
 	}

 	ack1.Ack()
 	if got, want := ackTracker.CommitOffset(), int64(2); got != want {
 		t.Errorf("ackTracker.CommitOffset() got %v, want %v", got, want)
 	}
 	if got, want := ackTracker.Empty(), false; got != want {
 		t.Errorf("ackTracker.Empty() got %v, want %v", got, want)
 	}

 	// Skipped ack2, so the commit offset should not have been updated.
 	ack3.Ack()
 	if got, want := ackTracker.CommitOffset(), int64(2); got != want {
 		t.Errorf("ackTracker.CommitOffset() got %v, want %v", got, want)
 	}
 	if got, want := ackTracker.Empty(), false; got != want {
 		t.Errorf("ackTracker.Empty() got %v, want %v", got, want)
 	}

 	// Both ack2 and ack3 should be removed from the outstanding acks queue.
 	ack2.Ack()
 	if got, want := ackTracker.CommitOffset(), int64(4); got != want {
 		t.Errorf("ackTracker.CommitOffset() got %v, want %v", got, want)
 	}
 	if got, want := ackTracker.Empty(), true; got != want {
 		t.Errorf("ackTracker.Empty() got %v, want %v", got, want)
 	}

 	// Newly received message.
 	ack4 := newAckConsumer(4, 0, emptyAckConsumer)
 	if err := ackTracker.Push(ack4); err != nil {
 		t.Errorf("ackTracker.Push() got err %v", err)
 	}
 	ack4.Ack()
 	if got, want := ackTracker.Empty(), false; got != want {
 		t.Errorf("ackTracker.Empty() got %v, want %v", got, want)
 	}
 	if got, want := ackTracker.CommitOffset(), int64(5); got != want {
 		t.Errorf("ackTracker.CommitOffset() got %v, want %v", got, want)
 	}
 	if got, want := ackTracker.Empty(), true; got != want {
 		t.Errorf("ackTracker.Empty() got %v, want %v", got, want)
 	}
 }

 func TestAckTrackerRelease(t *testing.T) {
 	ackTracker := newAckTracker()
 	onAckAfterRelease := func(ac *ackConsumer) {
 		t.Error("onAck should not be called")
 	}
 	ack1 := newAckConsumer(1, 0, emptyAckConsumer)
 	ack2 := newAckConsumer(2, 0, onAckAfterRelease)
 	ack3 := newAckConsumer(3, 0, onAckAfterRelease)
 	ack4 := newAckConsumer(4, 0, onAckAfterRelease)

 	if err := ackTracker.Push(ack1); err != nil {
 		t.Errorf("ackTracker.Push() got err %v", err)
 	}
 	if err := ackTracker.Push(ack2); err != nil {
 		t.Errorf("ackTracker.Push() got err %v", err)
 	}
 	if err := ackTracker.Push(ack3); err != nil {
 		t.Errorf("ackTracker.Push() got err %v", err)
 	}

 	// First ack is called before Release and should be processed.
 	ack1.Ack()

 	// After clearing outstanding acks, onAck should not be called.
 	ackTracker.Release()
 	ack2.Ack()
 	ack3.Ack()
 	if got, want := ackTracker.Empty(), true; got != want {
 		t.Errorf("ackTracker.Empty() got %v, want %v", got, want)
 	}

 	// New acks should be cleared and discarded.
 	if err := ackTracker.Push(ack4); err != nil {
 		t.Errorf("ackTracker.Push() got err %v", err)
 	}
 	ack4.Ack()

 	if got, want := ackTracker.Empty(), true; got != want {
 		t.Errorf("ackTracker.Empty() got %v, want %v", got, want)
 	}
 	if got, want := ackTracker.CommitOffset(), int64(2); got != want {
 		t.Errorf("ackTracker.CommitOffset() got %v, want %v", got, want)
 	}
 }

 func TestAckTrackerReset(t *testing.T) {
 	ackTracker := newAckTracker()
 	onAckAfterReset := func(ac *ackConsumer) {
 		t.Error("onAck should not be called")
 	}
 	ack1 := newAckConsumer(1, 0, emptyAckConsumer)
 	ack2 := newAckConsumer(2, 0, emptyAckConsumer)
 	ack3 := newAckConsumer(3, 0, onAckAfterReset)

 	if err := ackTracker.Push(ack1); err != nil {
 		t.Errorf("ackTracker.Push() got err %v", err)
 	}
 	if err := ackTracker.Push(ack2); err != nil {
 		t.Errorf("ackTracker.Push() got err %v", err)
 	}
 	if err := ackTracker.Push(ack3); err != nil {
 		t.Errorf("ackTracker.Push() got err %v", err)
 	}

 	// Ack tracker should not allow duplicate msg1.
 	if got, want := ackTracker.Push(ack1), errOutOfOrderMessages; !test.ErrorEqual(got, want) {
 		t.Errorf("ackTracker.Push() got err %v, want err %v", got, want)
 	}

 	// Ack 2 messages to advance the commit offset.
 	ack1.Ack()
 	ack2.Ack()
 	if got, want := ackTracker.CommitOffset(), int64(3); got != want {
 		t.Errorf("ackTracker.CommitOffset() got %v, want %v", got, want)
 	}
 	if got, want := ackTracker.Empty(), false; got != want {
 		t.Errorf("ackTracker.Empty() got %v, want %v", got, want)
 	}

 	// Reset should clear the outstanding acks and reset the desired commit
 	// offset.
 	ackTracker.Reset()
 	// Outstanding ack3 should be invalidated.
 	ack3.Ack()
 	if got, want := ackTracker.CommitOffset(), nilCursorOffset; got != want {
 		t.Errorf("ackTracker.CommitOffset() got %v, want %v", got, want)
 	}
 	if got, want := ackTracker.Empty(), true; got != want {
 		t.Errorf("ackTracker.Empty() got %v, want %v", got, want)
 	}

 	// After reset, msg1 should be accepted and processed.
 	ack1 = newAckConsumer(1, 0, emptyAckConsumer)
 	if err := ackTracker.Push(ack1); err != nil {
 		t.Errorf("ackTracker.Push() got err %v", err)
 	}
 	if got, want := ackTracker.Empty(), false; got != want {
 		t.Errorf("ackTracker.Empty() got %v, want %v", got, want)
 	}
 	ack1.Ack()
 	if got, want := ackTracker.CommitOffset(), int64(2); got != want {
 		t.Errorf("ackTracker.CommitOffset() got %v, want %v", got, want)
 	}
 }

 func TestCommitCursorTrackerProcessing(t *testing.T) {
 	ackTracker := newAckTracker()
 	commitTracker := newCommitCursorTracker(ackTracker)

 	// No messages received yet.
 	if got, want := commitTracker.NextOffset(), nilCursorOffset; got != want {
 		t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
 	}
 	if got, want := commitTracker.UpToDate(), true; got != want {
 		t.Errorf("commitCursorTracker.UpToDate() got %v, want %v", got, want)
 	}

 	ack1 := newAckConsumer(1, 0, emptyAckConsumer)
 	ack2 := newAckConsumer(2, 0, emptyAckConsumer)
 	ack3 := newAckConsumer(3, 0, emptyAckConsumer)
 	if err := ackTracker.Push(ack1); err != nil {
 		t.Errorf("ackTracker.Push() got err %v", err)
 	}
 	if err := ackTracker.Push(ack2); err != nil {
 		t.Errorf("ackTracker.Push() got err %v", err)
 	}
 	if err := ackTracker.Push(ack3); err != nil {
 		t.Errorf("ackTracker.Push() got err %v", err)
 	}

 	// All messages unacked.
 	if got, want := commitTracker.NextOffset(), nilCursorOffset; got != want {
 		t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
 	}
 	if got, want := commitTracker.UpToDate(), false; got != want {
 		t.Errorf("commitCursorTracker.UpToDate() got %v, want %v", got, want)
 	}

 	// Msg1 acked and commit sent to stream.
 	ack1.Ack()
 	if got, want := commitTracker.NextOffset(), int64(2); got != want {
 		t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
 	}
 	commitTracker.AddPending(commitTracker.NextOffset())
 	if got, want := commitTracker.NextOffset(), nilCursorOffset; got != want {
 		t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
 	}
 	if got, want := commitTracker.UpToDate(), false; got != want {
 		t.Errorf("commitCursorTracker.UpToDate() got %v, want %v", got, want)
 	}

 	// Msg 2 & 3 acked commit and sent to stream.
 	ack2.Ack()
 	if got, want := commitTracker.NextOffset(), int64(3); got != want {
 		t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
 	}
 	ack3.Ack()
 	if got, want := commitTracker.NextOffset(), int64(4); got != want {
 		t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
 	}
 	commitTracker.AddPending(commitTracker.NextOffset())
 	if got, want := commitTracker.NextOffset(), nilCursorOffset; got != want {
 		t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
 	}
 	if got, want := commitTracker.UpToDate(), false; got != want {
 		t.Errorf("commitCursorTracker.UpToDate() got %v, want %v", got, want)
 	}

 	// First 2 pending commits acknowledged.
 	if got, want := commitTracker.lastConfirmedOffset, nilCursorOffset; got != want {
 		t.Errorf("commitCursorTracker.lastConfirmedOffset got %v, want %v", got, want)
 	}
 	if err := commitTracker.ConfirmOffsets(2); err != nil {
 		t.Errorf("commitCursorTracker.ConfirmOffsets() got err %v", err)
 	}
 	if got, want := commitTracker.lastConfirmedOffset, int64(4); got != want {
 		t.Errorf("commitCursorTracker.lastConfirmedOffset got %v, want %v", got, want)
 	}
 	if got, want := commitTracker.UpToDate(), true; got != want {
 		t.Errorf("commitCursorTracker.UpToDate() got %v, want %v", got, want)
 	}
 }

 func TestCommitCursorTrackerReset(t *testing.T) {
 	ackTracker := newAckTracker()
 	commitTracker := newCommitCursorTracker(ackTracker)

 	onAckAfterReset := func(ac *ackConsumer) {
 		t.Error("onAck should not be called")
 	}
 	ack1 := newAckConsumer(1, 0, emptyAckConsumer)
 	ack2 := newAckConsumer(2, 0, emptyAckConsumer)
 	ack3 := newAckConsumer(3, 0, onAckAfterReset)
 	if err := ackTracker.Push(ack1); err != nil {
 		t.Errorf("ackTracker.Push() got err %v", err)
 	}
 	if err := ackTracker.Push(ack2); err != nil {
 		t.Errorf("ackTracker.Push() got err %v", err)
 	}
 	if err := ackTracker.Push(ack3); err != nil {
 		t.Errorf("ackTracker.Push() got err %v", err)
 	}

 	// Ack and commit 2 messages.
 	ack1.Ack()
 	ack2.Ack()
 	if got, want := commitTracker.NextOffset(), int64(3); got != want {
 		t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
 	}
 	commitTracker.AddPending(commitTracker.NextOffset())
 	if err := commitTracker.ConfirmOffsets(1); err != nil {
 		t.Errorf("commitCursorTracker.ConfirmOffsets() got err %v", err)
 	}

 	// Reset should clear the ack tracker and commit tracker to their initial
 	// states.
 	commitTracker.Reset()
 	// Outstanding ack3 should be invalidated.
 	ack3.Ack()
 	if got, want := ackTracker.CommitOffset(), nilCursorOffset; got != want {
 		t.Errorf("ackTracker.CommitOffset() got %v, want %v", got, want)
 	}
 	if got, want := ackTracker.Empty(), true; got != want {
 		t.Errorf("ackTracker.Empty() got %v, want %v", got, want)
 	}
 	if got, want := commitTracker.NextOffset(), nilCursorOffset; got != want {
 		t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
 	}
 	if got, want := commitTracker.UpToDate(), true; got != want {
 		t.Errorf("commitCursorTracker.UpToDate() got %v, want %v", got, want)
 	}

 	// After reset, msg1 should be accepted and processed.
 	ack1 = newAckConsumer(1, 0, emptyAckConsumer)
 	if err := ackTracker.Push(ack1); err != nil {
 		t.Errorf("ackTracker.Push() got err %v", err)
 	}
 	ack1.Ack()
 	if got, want := commitTracker.NextOffset(), int64(2); got != want {
 		t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
 	}
 	commitTracker.AddPending(commitTracker.NextOffset())
 	if err := commitTracker.ConfirmOffsets(1); err != nil {
 		t.Errorf("commitCursorTracker.ConfirmOffsets() got err %v", err)
 	}
 }

 func TestCommitCursorTrackerStreamReconnects(t *testing.T) {
 	ackTracker := newAckTracker()
 	commitTracker := newCommitCursorTracker(ackTracker)

 	ack1 := newAckConsumer(1, 0, emptyAckConsumer)
 	ack2 := newAckConsumer(2, 0, emptyAckConsumer)
 	ack3 := newAckConsumer(3, 0, emptyAckConsumer)
 	if err := ackTracker.Push(ack1); err != nil {
 		t.Errorf("ackTracker.Push() got err %v", err)
 	}
 	if err := ackTracker.Push(ack2); err != nil {
 		t.Errorf("ackTracker.Push() got err %v", err)
 	}
 	if err := ackTracker.Push(ack3); err != nil {
 		t.Errorf("ackTracker.Push() got err %v", err)
 	}

 	// All messages unacked.
 	if got, want := commitTracker.NextOffset(), nilCursorOffset; got != want {
 		t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
 	}
 	if got, want := commitTracker.UpToDate(), false; got != want {
 		t.Errorf("commitCursorTracker.UpToDate() got %v, want %v", got, want)
 	}

 	// Msg1 acked and commit sent to stream.
 	ack1.Ack()
 	if got, want := commitTracker.NextOffset(), int64(2); got != want {
 		t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
 	}
 	commitTracker.AddPending(commitTracker.NextOffset())
 	if got, want := commitTracker.NextOffset(), nilCursorOffset; got != want {
 		t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
 	}
 	if got, want := commitTracker.UpToDate(), false; got != want {
 		t.Errorf("commitCursorTracker.UpToDate() got %v, want %v", got, want)
 	}

 	// Msg2 acked and commit sent to stream.
 	ack2.Ack()
 	if got, want := commitTracker.NextOffset(), int64(3); got != want {
 		t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
 	}
 	commitTracker.AddPending(commitTracker.NextOffset())
 	if got, want := commitTracker.NextOffset(), nilCursorOffset; got != want {
 		t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
 	}
 	if got, want := commitTracker.UpToDate(), false; got != want {
 		t.Errorf("commitCursorTracker.UpToDate() got %v, want %v", got, want)
 	}

 	// Stream breaks and pending offsets are cleared.
 	commitTracker.ClearPending()
 	if got, want := commitTracker.UpToDate(), false; got != want {
 		t.Errorf("commitCursorTracker.UpToDate() got %v, want %v", got, want)
 	}
 	// When the stream reconnects the next offset should be 3 (offset 2 skipped).
 	if got, want := commitTracker.NextOffset(), int64(3); got != want {
 		t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
 	}
 	commitTracker.AddPending(commitTracker.NextOffset())
 	if got, want := commitTracker.NextOffset(), nilCursorOffset; got != want {
 		t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
 	}
 	if got, want := commitTracker.UpToDate(), false; got != want {
 		t.Errorf("commitCursorTracker.UpToDate() got %v, want %v", got, want)
 	}

 	// Msg3 acked and commit sent to stream.
 	ack3.Ack()
 	if got, want := commitTracker.NextOffset(), int64(4); got != want {
 		t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
 	}
 	commitTracker.AddPending(commitTracker.NextOffset())
 	if got, want := commitTracker.NextOffset(), nilCursorOffset; got != want {
 		t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
 	}
 	if got, want := commitTracker.UpToDate(), false; got != want {
 		t.Errorf("commitCursorTracker.UpToDate() got %v, want %v", got, want)
 	}

 	// Only 1 pending commit confirmed.
 	if got, want := commitTracker.lastConfirmedOffset, nilCursorOffset; got != want {
 		t.Errorf("commitCursorTracker.lastConfirmedOffset got %v, want %v", got, want)
 	}
 	if err := commitTracker.ConfirmOffsets(1); err != nil {
 		t.Errorf("commitCursorTracker.ConfirmOffsets() got err %v", err)
 	}
 	if got, want := commitTracker.lastConfirmedOffset, int64(3); got != want {
 		t.Errorf("commitCursorTracker.lastConfirmedOffset got %v, want %v", got, want)
 	}
 	if got, want := commitTracker.UpToDate(), false; got != want {
 		t.Errorf("commitCursorTracker.UpToDate() got %v, want %v", got, want)
 	}

 	// Final pending commit confirmed.
 	if err := commitTracker.ConfirmOffsets(1); err != nil {
 		t.Errorf("commitCursorTracker.ConfirmOffsets() got err %v", err)
 	}
 	if got, want := commitTracker.lastConfirmedOffset, int64(4); got != want {
 		t.Errorf("commitCursorTracker.lastConfirmedOffset got %v, want %v", got, want)
 	}
 	if got, want := commitTracker.UpToDate(), true; got != want {
 		t.Errorf("commitCursorTracker.UpToDate() got %v, want %v", got, want)
 	}
 }
	// Copyright 2020 Google LLC
	//
	// Licensed under the Apache License, Version 2.0 (the "License");
	// you may not use this file except in compliance with the License.
	// You may obtain a copy of the License at
	//
	// https://www.apache.org/licenses/LICENSE-2.0
	//
	// Unless required by applicable law or agreed to in writing, software
	// distributed under the License is distributed on an "AS IS" BASIS,
	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	// See the License for the specific language governing permissions and

	package wire

	import (
	"testing"

	"cloud.google.com/go/pubsublite/internal/test"
	)

	func emptyAckConsumer(_ *ackConsumer) {
	// Nothing to do.
	}

	func TestAckConsumerAck(t *testing.T) {
	numAcks := 0
	onAck := func(ac *ackConsumer) {
	numAcks++
	}
	ackConsumer := newAckConsumer(0, 0, onAck)
	if got, want := ackConsumer.IsAcked(), false; got != want {
	t.Errorf("ackConsumer.IsAcked() got %v, want %v", got, want)
	}

	// Test duplicate acks.
	for i := 0; i < 3; i++ {
	ackConsumer.Ack()

	if got, want := ackConsumer.IsAcked(), true; got != want {
	t.Errorf("ackConsumer.IsAcked() got %v, want %v", got, want)
	}
	if got, want := numAcks, 1; got != want {
	t.Errorf("onAck func called %v times, expected %v call", got, want)
	}
	}
	}

	func TestAckConsumerClear(t *testing.T) {
	onAck := func(ac *ackConsumer) {
	t.Error("onAck func should not have been called")
	}
	ackConsumer := newAckConsumer(0, 0, onAck)
	ackConsumer.Clear()
	ackConsumer.Ack()

	if got, want := ackConsumer.IsAcked(), true; got != want {
	t.Errorf("ackConsumer.IsAcked() got %v, want %v", got, want)
	}
	}

	func TestAckTrackerProcessing(t *testing.T) {
	ackTracker := newAckTracker()

	// No messages received yet.
	if got, want := ackTracker.CommitOffset(), nilCursorOffset; got != want {
	t.Errorf("ackTracker.CommitOffset() got %v, want %v", got, want)
	}

	ack1 := newAckConsumer(1, 0, emptyAckConsumer)
	ack2 := newAckConsumer(2, 0, emptyAckConsumer)
	ack3 := newAckConsumer(3, 0, emptyAckConsumer)
	if err := ackTracker.Push(ack1); err != nil {
	t.Errorf("ackTracker.Push() got err %v", err)
	}
	if err := ackTracker.Push(ack2); err != nil {
	t.Errorf("ackTracker.Push() got err %v", err)
	}
	if err := ackTracker.Push(ack3); err != nil {
	t.Errorf("ackTracker.Push() got err %v", err)
	}

	// All messages unacked.
	if got, want := ackTracker.CommitOffset(), nilCursorOffset; got != want {
	t.Errorf("ackTracker.CommitOffset() got %v, want %v", got, want)
	}

	ack1.Ack()
	if got, want := ackTracker.CommitOffset(), int64(2); got != want {
	t.Errorf("ackTracker.CommitOffset() got %v, want %v", got, want)
	}
	if got, want := ackTracker.Empty(), false; got != want {
	t.Errorf("ackTracker.Empty() got %v, want %v", got, want)
	}

	// Skipped ack2, so the commit offset should not have been updated.
	ack3.Ack()
	if got, want := ackTracker.CommitOffset(), int64(2); got != want {
	t.Errorf("ackTracker.CommitOffset() got %v, want %v", got, want)
	}
	if got, want := ackTracker.Empty(), false; got != want {
	t.Errorf("ackTracker.Empty() got %v, want %v", got, want)
	}

	// Both ack2 and ack3 should be removed from the outstanding acks queue.
	ack2.Ack()
	if got, want := ackTracker.CommitOffset(), int64(4); got != want {
	t.Errorf("ackTracker.CommitOffset() got %v, want %v", got, want)
	}
	if got, want := ackTracker.Empty(), true; got != want {
	t.Errorf("ackTracker.Empty() got %v, want %v", got, want)
	}

	// Newly received message.
	ack4 := newAckConsumer(4, 0, emptyAckConsumer)
	if err := ackTracker.Push(ack4); err != nil {
	t.Errorf("ackTracker.Push() got err %v", err)
	}
	ack4.Ack()
	if got, want := ackTracker.Empty(), false; got != want {
	t.Errorf("ackTracker.Empty() got %v, want %v", got, want)
	}
	if got, want := ackTracker.CommitOffset(), int64(5); got != want {
	t.Errorf("ackTracker.CommitOffset() got %v, want %v", got, want)
	}
	if got, want := ackTracker.Empty(), true; got != want {
	t.Errorf("ackTracker.Empty() got %v, want %v", got, want)
	}
	}

	func TestAckTrackerRelease(t *testing.T) {
	ackTracker := newAckTracker()
	onAckAfterRelease := func(ac *ackConsumer) {
	t.Error("onAck should not be called")
	}
	ack1 := newAckConsumer(1, 0, emptyAckConsumer)
	ack2 := newAckConsumer(2, 0, onAckAfterRelease)
	ack3 := newAckConsumer(3, 0, onAckAfterRelease)
	ack4 := newAckConsumer(4, 0, onAckAfterRelease)

	if err := ackTracker.Push(ack1); err != nil {
	t.Errorf("ackTracker.Push() got err %v", err)
	}
	if err := ackTracker.Push(ack2); err != nil {
	t.Errorf("ackTracker.Push() got err %v", err)
	}
	if err := ackTracker.Push(ack3); err != nil {
	t.Errorf("ackTracker.Push() got err %v", err)
	}

	// First ack is called before Release and should be processed.
	ack1.Ack()

	// After clearing outstanding acks, onAck should not be called.
	ackTracker.Release()
	ack2.Ack()
	ack3.Ack()
	if got, want := ackTracker.Empty(), true; got != want {
	t.Errorf("ackTracker.Empty() got %v, want %v", got, want)
	}

	// New acks should be cleared and discarded.
	if err := ackTracker.Push(ack4); err != nil {
	t.Errorf("ackTracker.Push() got err %v", err)
	}
	ack4.Ack()

	if got, want := ackTracker.Empty(), true; got != want {
	t.Errorf("ackTracker.Empty() got %v, want %v", got, want)
	}
	if got, want := ackTracker.CommitOffset(), int64(2); got != want {
	t.Errorf("ackTracker.CommitOffset() got %v, want %v", got, want)
	}
	}

	func TestAckTrackerReset(t *testing.T) {
	ackTracker := newAckTracker()
	onAckAfterReset := func(ac *ackConsumer) {
	t.Error("onAck should not be called")
	}
	ack1 := newAckConsumer(1, 0, emptyAckConsumer)
	ack2 := newAckConsumer(2, 0, emptyAckConsumer)
	ack3 := newAckConsumer(3, 0, onAckAfterReset)

	if err := ackTracker.Push(ack1); err != nil {
	t.Errorf("ackTracker.Push() got err %v", err)
	}
	if err := ackTracker.Push(ack2); err != nil {
	t.Errorf("ackTracker.Push() got err %v", err)
	}
	if err := ackTracker.Push(ack3); err != nil {
	t.Errorf("ackTracker.Push() got err %v", err)
	}

	// Ack tracker should not allow duplicate msg1.
	if got, want := ackTracker.Push(ack1), errOutOfOrderMessages; !test.ErrorEqual(got, want) {
	t.Errorf("ackTracker.Push() got err %v, want err %v", got, want)
	}

	// Ack 2 messages to advance the commit offset.
	ack1.Ack()
	ack2.Ack()
	if got, want := ackTracker.CommitOffset(), int64(3); got != want {
	t.Errorf("ackTracker.CommitOffset() got %v, want %v", got, want)
	}
	if got, want := ackTracker.Empty(), false; got != want {
	t.Errorf("ackTracker.Empty() got %v, want %v", got, want)
	}

	// Reset should clear the outstanding acks and reset the desired commit
	// offset.
	ackTracker.Reset()
	// Outstanding ack3 should be invalidated.
	ack3.Ack()
	if got, want := ackTracker.CommitOffset(), nilCursorOffset; got != want {
	t.Errorf("ackTracker.CommitOffset() got %v, want %v", got, want)
	}
	if got, want := ackTracker.Empty(), true; got != want {
	t.Errorf("ackTracker.Empty() got %v, want %v", got, want)
	}

	// After reset, msg1 should be accepted and processed.
	ack1 = newAckConsumer(1, 0, emptyAckConsumer)
	if err := ackTracker.Push(ack1); err != nil {
	t.Errorf("ackTracker.Push() got err %v", err)
	}
	if got, want := ackTracker.Empty(), false; got != want {
	t.Errorf("ackTracker.Empty() got %v, want %v", got, want)
	}
	ack1.Ack()
	if got, want := ackTracker.CommitOffset(), int64(2); got != want {
	t.Errorf("ackTracker.CommitOffset() got %v, want %v", got, want)
	}
	}

	func TestCommitCursorTrackerProcessing(t *testing.T) {
	ackTracker := newAckTracker()
	commitTracker := newCommitCursorTracker(ackTracker)

	// No messages received yet.
	if got, want := commitTracker.NextOffset(), nilCursorOffset; got != want {
	t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
	}
	if got, want := commitTracker.UpToDate(), true; got != want {
	t.Errorf("commitCursorTracker.UpToDate() got %v, want %v", got, want)
	}

	ack1 := newAckConsumer(1, 0, emptyAckConsumer)
	ack2 := newAckConsumer(2, 0, emptyAckConsumer)
	ack3 := newAckConsumer(3, 0, emptyAckConsumer)
	if err := ackTracker.Push(ack1); err != nil {
	t.Errorf("ackTracker.Push() got err %v", err)
	}
	if err := ackTracker.Push(ack2); err != nil {
	t.Errorf("ackTracker.Push() got err %v", err)
	}
	if err := ackTracker.Push(ack3); err != nil {
	t.Errorf("ackTracker.Push() got err %v", err)
	}

	// All messages unacked.
	if got, want := commitTracker.NextOffset(), nilCursorOffset; got != want {
	t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
	}
	if got, want := commitTracker.UpToDate(), false; got != want {
	t.Errorf("commitCursorTracker.UpToDate() got %v, want %v", got, want)
	}

	// Msg1 acked and commit sent to stream.
	ack1.Ack()
	if got, want := commitTracker.NextOffset(), int64(2); got != want {
	t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
	}
	commitTracker.AddPending(commitTracker.NextOffset())
	if got, want := commitTracker.NextOffset(), nilCursorOffset; got != want {
	t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
	}
	if got, want := commitTracker.UpToDate(), false; got != want {
	t.Errorf("commitCursorTracker.UpToDate() got %v, want %v", got, want)
	}

	// Msg 2 & 3 acked commit and sent to stream.
	ack2.Ack()
	if got, want := commitTracker.NextOffset(), int64(3); got != want {
	t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
	}
	ack3.Ack()
	if got, want := commitTracker.NextOffset(), int64(4); got != want {
	t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
	}
	commitTracker.AddPending(commitTracker.NextOffset())
	if got, want := commitTracker.NextOffset(), nilCursorOffset; got != want {
	t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
	}
	if got, want := commitTracker.UpToDate(), false; got != want {
	t.Errorf("commitCursorTracker.UpToDate() got %v, want %v", got, want)
	}

	// First 2 pending commits acknowledged.
	if got, want := commitTracker.lastConfirmedOffset, nilCursorOffset; got != want {
	t.Errorf("commitCursorTracker.lastConfirmedOffset got %v, want %v", got, want)
	}
	if err := commitTracker.ConfirmOffsets(2); err != nil {
	t.Errorf("commitCursorTracker.ConfirmOffsets() got err %v", err)
	}
	if got, want := commitTracker.lastConfirmedOffset, int64(4); got != want {
	t.Errorf("commitCursorTracker.lastConfirmedOffset got %v, want %v", got, want)
	}
	if got, want := commitTracker.UpToDate(), true; got != want {
	t.Errorf("commitCursorTracker.UpToDate() got %v, want %v", got, want)
	}
	}

	func TestCommitCursorTrackerReset(t *testing.T) {
	ackTracker := newAckTracker()
	commitTracker := newCommitCursorTracker(ackTracker)

	onAckAfterReset := func(ac *ackConsumer) {
	t.Error("onAck should not be called")
	}
	ack1 := newAckConsumer(1, 0, emptyAckConsumer)
	ack2 := newAckConsumer(2, 0, emptyAckConsumer)
	ack3 := newAckConsumer(3, 0, onAckAfterReset)
	if err := ackTracker.Push(ack1); err != nil {
	t.Errorf("ackTracker.Push() got err %v", err)
	}
	if err := ackTracker.Push(ack2); err != nil {
	t.Errorf("ackTracker.Push() got err %v", err)
	}
	if err := ackTracker.Push(ack3); err != nil {
	t.Errorf("ackTracker.Push() got err %v", err)
	}

	// Ack and commit 2 messages.
	ack1.Ack()
	ack2.Ack()
	if got, want := commitTracker.NextOffset(), int64(3); got != want {
	t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
	}
	commitTracker.AddPending(commitTracker.NextOffset())
	if err := commitTracker.ConfirmOffsets(1); err != nil {
	t.Errorf("commitCursorTracker.ConfirmOffsets() got err %v", err)
	}

	// Reset should clear the ack tracker and commit tracker to their initial
	// states.
	commitTracker.Reset()
	// Outstanding ack3 should be invalidated.
	ack3.Ack()
	if got, want := ackTracker.CommitOffset(), nilCursorOffset; got != want {
	t.Errorf("ackTracker.CommitOffset() got %v, want %v", got, want)
	}
	if got, want := ackTracker.Empty(), true; got != want {
	t.Errorf("ackTracker.Empty() got %v, want %v", got, want)
	}
	if got, want := commitTracker.NextOffset(), nilCursorOffset; got != want {
	t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
	}
	if got, want := commitTracker.UpToDate(), true; got != want {
	t.Errorf("commitCursorTracker.UpToDate() got %v, want %v", got, want)
	}

	// After reset, msg1 should be accepted and processed.
	ack1 = newAckConsumer(1, 0, emptyAckConsumer)
	if err := ackTracker.Push(ack1); err != nil {
	t.Errorf("ackTracker.Push() got err %v", err)
	}
	ack1.Ack()
	if got, want := commitTracker.NextOffset(), int64(2); got != want {
	t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
	}
	commitTracker.AddPending(commitTracker.NextOffset())
	if err := commitTracker.ConfirmOffsets(1); err != nil {
	t.Errorf("commitCursorTracker.ConfirmOffsets() got err %v", err)
	}
	}

	func TestCommitCursorTrackerStreamReconnects(t *testing.T) {
	ackTracker := newAckTracker()
	commitTracker := newCommitCursorTracker(ackTracker)

	ack1 := newAckConsumer(1, 0, emptyAckConsumer)
	ack2 := newAckConsumer(2, 0, emptyAckConsumer)
	ack3 := newAckConsumer(3, 0, emptyAckConsumer)
	if err := ackTracker.Push(ack1); err != nil {
	t.Errorf("ackTracker.Push() got err %v", err)
	}
	if err := ackTracker.Push(ack2); err != nil {
	t.Errorf("ackTracker.Push() got err %v", err)
	}
	if err := ackTracker.Push(ack3); err != nil {
	t.Errorf("ackTracker.Push() got err %v", err)
	}

	// All messages unacked.
	if got, want := commitTracker.NextOffset(), nilCursorOffset; got != want {
	t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
	}
	if got, want := commitTracker.UpToDate(), false; got != want {
	t.Errorf("commitCursorTracker.UpToDate() got %v, want %v", got, want)
	}

	// Msg1 acked and commit sent to stream.
	ack1.Ack()
	if got, want := commitTracker.NextOffset(), int64(2); got != want {
	t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
	}
	commitTracker.AddPending(commitTracker.NextOffset())
	if got, want := commitTracker.NextOffset(), nilCursorOffset; got != want {
	t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
	}
	if got, want := commitTracker.UpToDate(), false; got != want {
	t.Errorf("commitCursorTracker.UpToDate() got %v, want %v", got, want)
	}

	// Msg2 acked and commit sent to stream.
	ack2.Ack()
	if got, want := commitTracker.NextOffset(), int64(3); got != want {
	t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
	}
	commitTracker.AddPending(commitTracker.NextOffset())
	if got, want := commitTracker.NextOffset(), nilCursorOffset; got != want {
	t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
	}
	if got, want := commitTracker.UpToDate(), false; got != want {
	t.Errorf("commitCursorTracker.UpToDate() got %v, want %v", got, want)
	}

	// Stream breaks and pending offsets are cleared.
	commitTracker.ClearPending()
	if got, want := commitTracker.UpToDate(), false; got != want {
	t.Errorf("commitCursorTracker.UpToDate() got %v, want %v", got, want)
	}
	// When the stream reconnects the next offset should be 3 (offset 2 skipped).
	if got, want := commitTracker.NextOffset(), int64(3); got != want {
	t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
	}
	commitTracker.AddPending(commitTracker.NextOffset())
	if got, want := commitTracker.NextOffset(), nilCursorOffset; got != want {
	t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
	}
	if got, want := commitTracker.UpToDate(), false; got != want {
	t.Errorf("commitCursorTracker.UpToDate() got %v, want %v", got, want)
	}

	// Msg3 acked and commit sent to stream.
	ack3.Ack()
	if got, want := commitTracker.NextOffset(), int64(4); got != want {
	t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
	}
	commitTracker.AddPending(commitTracker.NextOffset())
	if got, want := commitTracker.NextOffset(), nilCursorOffset; got != want {
	t.Errorf("commitCursorTracker.NextOffset() got %v, want %v", got, want)
	}
	if got, want := commitTracker.UpToDate(), false; got != want {
	t.Errorf("commitCursorTracker.UpToDate() got %v, want %v", got, want)
	}

	// Only 1 pending commit confirmed.
	if got, want := commitTracker.lastConfirmedOffset, nilCursorOffset; got != want {
	t.Errorf("commitCursorTracker.lastConfirmedOffset got %v, want %v", got, want)
	}
	if err := commitTracker.ConfirmOffsets(1); err != nil {
	t.Errorf("commitCursorTracker.ConfirmOffsets() got err %v", err)
	}
	if got, want := commitTracker.lastConfirmedOffset, int64(3); got != want {
	t.Errorf("commitCursorTracker.lastConfirmedOffset got %v, want %v", got, want)
	}
	if got, want := commitTracker.UpToDate(), false; got != want {
	t.Errorf("commitCursorTracker.UpToDate() got %v, want %v", got, want)
	}

	// Final pending commit confirmed.
	if err := commitTracker.ConfirmOffsets(1); err != nil {
	t.Errorf("commitCursorTracker.ConfirmOffsets() got err %v", err)
	}
	if got, want := commitTracker.lastConfirmedOffset, int64(4); got != want {
	t.Errorf("commitCursorTracker.lastConfirmedOffset got %v, want %v", got, want)
	}
	if got, want := commitTracker.UpToDate(), true; got != want {
	t.Errorf("commitCursorTracker.UpToDate() got %v, want %v", got, want)
	}
	}