Remove TestConsumerGroupDurableConsumer
@@ -398,84 +398,6 @@ func TestConsumerGroupLifecycle(t *testing.T) {
require.Nil(t, s1.metadata.GetConsumerGroup(group))
}
-// Ensure consumer groups track position in a partition so a consumer can
-// resume where it left off.
-func TestConsumerGroupDurableConsumer(t *testing.T) {
- defer cleanupStorage(t)
-
- // Configure server.
- s1Config := getTestConfig("a", true, 5050)
- s1Config.CursorsStream.Partitions = 1
- s1 := runServerWithConfig(t, s1Config)
- defer s1.Stop()
-
- getMetadataLeader(t, 10*time.Second, s1)
-
- client, err := lift.Connect([]string{"localhost:5050"})
- require.NoError(t, err)
- defer client.Close()
-
- // Create streams.
- err = client.CreateStream(context.Background(), "foo", "foo")
- require.NoError(t, err)
-
- // Publish some messages.
- for i := 0; i < 5; i++ {
- _, err = client.Publish(context.Background(), "foo", []byte(strconv.Itoa(i)))
- require.NoError(t, err)
- }
-
- group := "my-group"
-
- // Create consumer.
- cons, err := client.CreateConsumer(group, lift.AutoCheckpoint(0), lift.AutoOffsetEarliest())
- require.NoError(t, err)
-
- ch := make(chan *lift.Message)
-
- err = cons.Subscribe(context.Background(), []string{"foo"}, func(msg *lift.Message, err error) {
- require.NoError(t, err)
- require.NoError(t, cons.Checkpoint(context.Background()))
- ch <- msg
- })
- require.NoError(t, err)
-
- offset := int64(0)
- for i := 0; i < 3; i++ {
- select {
- case msg := <-ch:
- require.Equal(t, int64(i), msg.Offset())
- offset = msg.Offset()
- case <-time.After(5 * time.Second):
- t.Fatal("Did not receive expected message")
- }
- }
-
- // Close consumer.
- require.NoError(t, cons.Close())
-
- // Create consumer that should resume at offset 3.
- cons, err = client.CreateConsumer(group, lift.AutoCheckpoint(0))
- require.NoError(t, err)
-
- ch = make(chan *lift.Message)
-
- err = cons.Subscribe(context.Background(), []string{"foo"}, func(msg *lift.Message, err error) {
- require.NoError(t, err)
- ch <- msg
- })
- require.NoError(t, err)
-
- for i := 0; i < 2; i++ {
- select {
- case msg := <-ch:
- require.Equal(t, int64(offset+int64(i)+1), msg.Offset())
- case <-time.After(5 * time.Second):
- t.Fatal("Did not receive expected message")
- }
- }
-}
-
// Ensure joining a consumer group fails if the consumer subscribes to a stream
// that doesn't exist or if a consumer with the same ID is already a member.
func TestConsumerGroupAddMember(t *testing.T) {
