feat(go): add stream/topic purge API with BDD tests and fix server purge bug

bdd/docker-compose.yml

@@ -30,13 +30,15 @@ services:
     volumes:
       - ./scenarios/basic_messaging.feature:/app/features/basic_messaging.feature
       - ./scenarios/leader_redirection.feature:/app/features/leader_redirection.feature
+      - ./scenarios/stream_topic_purge.feature:/app/features/stream_topic_purge.feature
     command:
       - sh
       - -c
       - |
         case "$$BDD_FEATURE" in
           basic_messaging)    cargo test -p bdd --features bdd --test basic_messaging ;;
           leader_redirection) cargo test -p bdd --features bdd --test leader_redirection ;;
+          stream_topic_purge) cargo test -p bdd --features bdd --test stream_topic_purge ;;
           *)                  cargo test -p bdd --features bdd ;;
         esac
     networks:

bdd/go/tests/basic_messaging.go

@@ -21,13 +21,9 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"os"
 
-	"github.com/apache/iggy/foreign/go/client"
-	"github.com/apache/iggy/foreign/go/client/tcp"
 	iggcon "github.com/apache/iggy/foreign/go/contracts"
 	"github.com/cucumber/godog"
-	"github.com/google/uuid"
 )
 
 type basicMessagingCtxKey struct{}
@@ -52,36 +48,21 @@ type basicMessagingSteps struct{}
 
 func (s basicMessagingSteps) givenRunningServer(ctx context.Context) error {
 	c := getBasicMessagingCtx(ctx)
-	addr := os.Getenv("IGGY_TCP_ADDRESS")
-	if addr == "" {
-		addr = "127.0.0.1:8090"
-	}
+	addr := defaultServerAddress()
 	c.serverAddr = &addr
 	return nil
 }
 
 func (s basicMessagingSteps) givenAuthenticationAsRoot(ctx context.Context) error {
 	c := getBasicMessagingCtx(ctx)
-	serverAddr := *c.serverAddr
 
-	cli, err := client.NewIggyClient(
-		client.WithTcp(
-			tcp.WithServerAddress(serverAddr),
-		),
-	)
+	cli, err := connectToServer(ctx, *c.serverAddr)
 	if err != nil {
-		return fmt.Errorf("error creating client: %w", err)
-	}
-
-	if err = cli.Connect(ctx); err != nil {
-		return fmt.Errorf("error connecting to server: %w", err)
-	}
-	if err = cli.Ping(ctx); err != nil {
-		return fmt.Errorf("error pinging client: %w", err)
+		return fmt.Errorf("failed to connect to server: %w", err)
 	}
 
-	if _, err = cli.LoginUser(ctx, "iggy", "iggy"); err != nil {
-		return fmt.Errorf("error logging in: %v", err)
+	if err := loginAsRoot(ctx, cli); err != nil {
+		return fmt.Errorf("failed to login as root: %w", err)
 	}
 
 	c.client = cli
@@ -95,7 +76,7 @@ func (s basicMessagingSteps) whenSendMessages(
 	topicID uint32,
 	partitionID uint32) error {
 	c := getBasicMessagingCtx(ctx)
-	messages, err := s.createTestMessages(messagesCount)
+	messages, err := createTestMessages(messagesCount)
 	if err != nil {
 		return fmt.Errorf("error creating test messages: %w", err)
 	}
@@ -111,20 +92,6 @@ func (s basicMessagingSteps) whenSendMessages(
 	return nil
 }
 
-func (s basicMessagingSteps) createTestMessages(count uint32) ([]iggcon.IggyMessage, error) {
-	messages := make([]iggcon.IggyMessage, 0, count)
-	for i := 0; uint32(i) < count; i++ {
-		id := uuid.New()
-		payload := []byte(fmt.Sprintf("test message %d", i))
-		message, err := iggcon.NewIggyMessage(payload, iggcon.WithID(id))
-		if err != nil {
-			return nil, fmt.Errorf("failed to create message: %w", err)
-		}
-		messages = append(messages, message)
-	}
-	return messages, nil
-}
-
 func (s basicMessagingSteps) whenPollMessages(
 	ctx context.Context,
 	streamID uint32,

bdd/go/tests/helper.go

@@ -0,0 +1,129 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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
+//
+//   http://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 limitations
+// under the License.
+
+package tests
+
+import (
+	"context"
+	"fmt"
+	"os"
+
+	"github.com/apache/iggy/foreign/go/client"
+	"github.com/apache/iggy/foreign/go/client/tcp"
+	iggcon "github.com/apache/iggy/foreign/go/contracts"
+	"github.com/google/uuid"
+)
+
+func defaultServerAddress() string {
+	if addr := os.Getenv("IGGY_TCP_ADDRESS"); addr != "" {
+		return addr
+	}
+	return "127.0.0.1:8090"
+}
+
+func connectToServer(ctx context.Context, addr string) (iggcon.Client, error) {
+	cli, err := client.NewIggyClient(
+		client.WithTcp(
+			tcp.WithServerAddress(addr),
+		),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("error creating client: %w", err)
+	}
+	if err = cli.Connect(ctx); err != nil {
+		return nil, fmt.Errorf("error connecting to server: %w", err)
+	}
+	if err = cli.Ping(ctx); err != nil {
+		return nil, fmt.Errorf("error pinging server: %w", err)
+	}
+	return cli, nil
+}
+
+func loginAsRoot(ctx context.Context, cli iggcon.Client) error {
+	if _, err := cli.LoginUser(ctx, "iggy", "iggy"); err != nil {
+		return fmt.Errorf("error logging in: %w", err)
+	}
+	return nil
+}
+
+func createTestMessages(count uint32) ([]iggcon.IggyMessage, error) {
+	messages := make([]iggcon.IggyMessage, 0, count)
+	for i := range count {
+		id := uuid.New()
+		payload := fmt.Appendf(nil, "test message %d", i)
+		message, err := iggcon.NewIggyMessage(payload, iggcon.WithID(id))
+		if err != nil {
+			return nil, fmt.Errorf("failed to create message: %w", err)
+		}
+		messages = append(messages, message)
+	}
+	return messages, nil
+}
+
+func sendTestMessages(
+	ctx context.Context,
+	cli iggcon.Client,
+	streamID iggcon.Identifier,
+	topicID iggcon.Identifier,
+	partitionID uint32,
+	count uint32,
+) (*iggcon.IggyMessage, error) {
+	messages, err := createTestMessages(count)
+	if err != nil {
+		return nil, err
+	}
+	partitioning := iggcon.PartitionId(partitionID)
+	if err = cli.SendMessages(ctx, streamID, topicID, partitioning, messages); err != nil {
+		return nil, fmt.Errorf("failed to send messages: %w", err)
+	}
+	last := messages[len(messages)-1]
+	return &last, nil
+}
+
+func pollTestMessages(
+	ctx context.Context,
+	cli iggcon.Client,
+	streamID iggcon.Identifier,
+	topicID iggcon.Identifier,
+	partitionID uint32,
+	startOffset uint64,
+	maxCount uint32,
+) (*iggcon.PolledMessage, error) {
+	consumer := iggcon.DefaultConsumer()
+	polled, err := cli.PollMessages(
+		ctx,
+		streamID,
+		topicID,
+		consumer,
+		iggcon.OffsetPollingStrategy(startOffset),
+		maxCount,
+		false,
+		&partitionID,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to poll messages: %w", err)
+	}
+	return polled, nil
+}
+
+func assertMessageCount(polled *iggcon.PolledMessage, expected uint32) error {
+	actual := uint32(len(polled.Messages))
+	if actual != expected {
+		return fmt.Errorf("expected %d messages, got %d", expected, actual)
+	}
+	return nil
+}

bdd/go/tests/leader_redirection.go

@@ -27,8 +27,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/apache/iggy/foreign/go/client"
-	"github.com/apache/iggy/foreign/go/client/tcp"
 	iggcon "github.com/apache/iggy/foreign/go/contracts"
 	ierror "github.com/apache/iggy/foreign/go/errors"
 	"github.com/cucumber/godog"
@@ -69,21 +67,6 @@ func getLeaderContext(ctx context.Context) *leaderCtx {
 	return ctx.Value(leaderCtxKey{}).(*leaderCtx)
 }
 
-func createAndConnectClient(ctx context.Context, addr string) (iggcon.Client, error) {
-	cli, err := client.NewIggyClient(
-		client.WithTcp(
-			tcp.WithServerAddress(addr),
-		),
-	)
-	if err != nil {
-		return nil, err
-	}
-	if err = cli.Connect(ctx); err != nil {
-		return nil, err
-	}
-	return cli, nil
-}
-
 func resolveServerAddress(role string, port uint16) string {
 	role = strings.ToLower(role)
 
@@ -247,7 +230,7 @@ func (s leaderSteps) whenCreateClientToRole(ctx context.Context, role string, _
 		return fmt.Errorf("%s server should be configured", role)
 	}
 
-	cli, err := createAndConnectClient(ctx, addr)
+	cli, err := connectToServer(ctx, addr)
 	if err != nil {
 		return err
 	}
@@ -270,7 +253,7 @@ func (s leaderSteps) whenCreateClientDirectToLeader(ctx context.Context, port ui
 		return fmt.Errorf("leader should be on port %d, but address is %s", port, addr)
 	}
 	var err error
-	c.Clients["main"], err = createAndConnectClient(ctx, addr)
+	c.Clients["main"], err = connectToServer(ctx, addr)
 	if err != nil {
 		return err
 	}
@@ -287,7 +270,7 @@ func (s leaderSteps) whenCreateClientToPort(ctx context.Context, port uint16) er
 		return fmt.Errorf("server on port %d should be configured", port)
 	}
 	var err error
-	c.Clients["main"], err = createAndConnectClient(ctx, addr)
+	c.Clients["main"], err = connectToServer(ctx, addr)
 	if err != nil {
 		return err
 	}
@@ -302,7 +285,7 @@ func (s leaderSteps) whenCreateNamedClient(ctx context.Context, name string, por
 		return fmt.Errorf("server on port %d should be configured", port)
 	}
 	var err error
-	c.Clients[name], err = createAndConnectClient(ctx, addr)
+	c.Clients[name], err = connectToServer(ctx, addr)
 	if err != nil {
 		return err
 	}