From c3efcff95d21bd45a3be225fe0be0f2848e5d0a3 Mon Sep 17 00:00:00 2001 From: chengxi Date: Sat, 4 Jul 2026 11:24:06 -0400 Subject: [PATCH 1/6] feat(go): implement purge stream/topic function --- bdd/go/tests/suite_test.go | 10 ++++++++++ foreign/go/client/tcp/tcp_stream_management.go | 5 +++++ foreign/go/client/tcp/tcp_topic_management.go | 5 +++++ foreign/go/contracts/client.go | 8 ++++++++ foreign/go/internal/command/code.go | 2 ++ foreign/go/internal/command/stream.go | 12 ++++++++++++ foreign/go/internal/command/topic.go | 13 +++++++++++++ 7 files changed, 55 insertions(+) diff --git a/bdd/go/tests/suite_test.go b/bdd/go/tests/suite_test.go index 15d951b444..6462338f5a 100644 --- a/bdd/go/tests/suite_test.go +++ b/bdd/go/tests/suite_test.go @@ -51,6 +51,16 @@ func TestFeatures(t *testing.T) { }, }) } + if feature == "all" || feature == "stream_topic_purge" { + suites = append(suites, godog.TestSuite{ + ScenarioInitializer: initPurgeScenario, + Options: &godog.Options{ + Format: "pretty", + Paths: []string{"../../scenarios/stream_topic_purge.feature"}, + TestingT: t, + }, + }) + } if len(suites) == 0 { t.Fatalf("unknown BDD_FEATURE=%q", feature) diff --git a/foreign/go/client/tcp/tcp_stream_management.go b/foreign/go/client/tcp/tcp_stream_management.go index 6a8e27f9b7..eabd565034 100644 --- a/foreign/go/client/tcp/tcp_stream_management.go +++ b/foreign/go/client/tcp/tcp_stream_management.go @@ -82,3 +82,8 @@ func (c *IggyTcpClient) DeleteStream(ctx context.Context, id iggcon.Identifier) _, err := c.do(ctx, &command.DeleteStream{StreamId: id}) return err } + +func (c *IggyTcpClient) PurgeStream(ctx context.Context, streamId iggcon.Identifier) error { + _, err := c.do(ctx, &command.PurgeStream{StreamId: streamId}) + return err +} diff --git a/foreign/go/client/tcp/tcp_topic_management.go b/foreign/go/client/tcp/tcp_topic_management.go index bfe0ccfee1..bedfff4f1b 100644 --- a/foreign/go/client/tcp/tcp_topic_management.go +++ b/foreign/go/client/tcp/tcp_topic_management.go @@ -120,3 +120,8 @@ func (c *IggyTcpClient) DeleteTopic(ctx context.Context, streamId, topicId iggco _, err := c.do(ctx, &command.DeleteTopic{StreamId: streamId, TopicId: topicId}) return err } + +func (c *IggyTcpClient) PurgeTopic(ctx context.Context, streamId, topicId iggcon.Identifier) error { + _, err := c.do(ctx, &command.PurgeTopic{StreamId: streamId, TopicId: topicId}) + return err +} diff --git a/foreign/go/contracts/client.go b/foreign/go/contracts/client.go index 144cc17e3e..88f18050a1 100644 --- a/foreign/go/contracts/client.go +++ b/foreign/go/contracts/client.go @@ -53,6 +53,10 @@ type Client interface { // Authentication is required, and the permission to manage the streams. DeleteStream(ctx context.Context, id Identifier) error + // PurgeStream purge a stream by unique ID or name. + // Authentication is required, and the permission to manage the streams. + PurgeStream(ctx context.Context, streamId Identifier) error + // GetTopic Get the info about a specific topic by unique ID or name. // Authentication is required, and the permission to read the topics. GetTopic(ctx context.Context, streamId, topicId Identifier) (*TopicDetails, error) @@ -91,6 +95,10 @@ type Client interface { // Authentication is required, and the permission to manage the topics. DeleteTopic(ctx context.Context, streamId, topicId Identifier) error + // PurgeTopic purge a topic by unique stream and topic IDs or names. + // Authentication is required, and the permission to manage the topics. + PurgeTopic(ctx context.Context, streamId, topicId Identifier) error + // SendMessages sends messages using specified partitioning strategy to the given stream and topic by unique IDs or names. // Authentication is required, and the permission to send the messages. SendMessages( diff --git a/foreign/go/internal/command/code.go b/foreign/go/internal/command/code.go index 644abf2dd0..b8dc28741e 100644 --- a/foreign/go/internal/command/code.go +++ b/foreign/go/internal/command/code.go @@ -50,11 +50,13 @@ const ( CreateStreamCode Code = 202 DeleteStreamCode Code = 203 UpdateStreamCode Code = 204 + PurgeStreamCode Code = 205 GetTopicCode Code = 300 GetTopicsCode Code = 301 CreateTopicCode Code = 302 DeleteTopicCode Code = 303 UpdateTopicCode Code = 304 + PurgeTopicCode Code = 305 CreatePartitionsCode Code = 402 DeletePartitionsCode Code = 403 DeleteSegmentsCode Code = 503 diff --git a/foreign/go/internal/command/stream.go b/foreign/go/internal/command/stream.go index d3c3c4098f..706a7f78f8 100644 --- a/foreign/go/internal/command/stream.go +++ b/foreign/go/internal/command/stream.go @@ -96,3 +96,15 @@ func (d *DeleteStream) Code() Code { func (d *DeleteStream) MarshalBinary() ([]byte, error) { return d.StreamId.MarshalBinary() } + +type PurgeStream struct { + StreamId iggcon.Identifier +} + +func (p *PurgeStream) Code() Code { + return PurgeStreamCode +} + +func (p *PurgeStream) MarshalBinary() ([]byte, error) { + return p.StreamId.MarshalBinary() +} diff --git a/foreign/go/internal/command/topic.go b/foreign/go/internal/command/topic.go index 1ce8a988d0..76d1a6968d 100644 --- a/foreign/go/internal/command/topic.go +++ b/foreign/go/internal/command/topic.go @@ -130,6 +130,19 @@ func (d *DeleteTopic) MarshalBinary() ([]byte, error) { return iggcon.MarshalIdentifiers(d.StreamId, d.TopicId) } +type PurgeTopic struct { + StreamId iggcon.Identifier + TopicId iggcon.Identifier +} + +func (p *PurgeTopic) Code() Code { + return PurgeTopicCode +} + +func (p *PurgeTopic) MarshalBinary() ([]byte, error) { + return iggcon.MarshalIdentifiers(p.StreamId, p.TopicId) +} + type UpdateTopic struct { StreamId iggcon.Identifier `json:"streamId"` TopicId iggcon.Identifier `json:"topicId"` From c410303a549013b0625c3bdb465ae69f8bcb6a32 Mon Sep 17 00:00:00 2001 From: chengxi Date: Sat, 4 Jul 2026 13:56:02 -0400 Subject: [PATCH 2/6] test(sdk): add stream_topic_purge scenario --- bdd/scenarios/stream_topic_purge.feature | 76 ++++++++++++++++++++++++ 1 file changed, 76 insertions(+) create mode 100644 bdd/scenarios/stream_topic_purge.feature diff --git a/bdd/scenarios/stream_topic_purge.feature b/bdd/scenarios/stream_topic_purge.feature new file mode 100644 index 0000000000..3867e54cc2 --- /dev/null +++ b/bdd/scenarios/stream_topic_purge.feature @@ -0,0 +1,76 @@ +# 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. + +@stream-topic-purge +Feature: Stream and Topic Purge + As a developer using Apache Iggy + I want to purge all messages from a stream or topic + So that I can clear data without deleting the resource + + Background: + Given I have a running Iggy server + And I am authenticated as the root user + + Scenario: Purge stream removes all messages + Given I have a stream with name "purge-test-stream" + And I have a topic with name "purge-topic" in stream "purge-test-stream" with 3 partitions + And I have sent 10 messages to stream "purge-test-stream", topic "purge-topic", partition 0 + When I purge the stream "purge-test-stream" + And I poll messages from stream "purge-test-stream", topic "purge-topic", partition 0 starting from offset 0 + Then I should receive 0 messages + + Scenario: Purge topic removes all messages + Given I have a stream with name "purge-topic-stream" + And I have a topic with name "purge-target-topic" in stream "purge-topic-stream" with 3 partitions + And I have sent 10 messages to stream "purge-topic-stream", topic "purge-target-topic", partition 0 + When I purge topic "purge-target-topic" in stream "purge-topic-stream" + And I poll messages from stream "purge-topic-stream", topic "purge-target-topic", partition 0 starting from offset 0 + Then I should receive 0 messages + + Scenario: Purge stream does not affect other stream + Given I have a stream with name "purge-stream-a" + And I have a topic with name "topic-in-a" in stream "purge-stream-a" with 1 partition + And I have sent 10 messages to stream "purge-stream-a", topic "topic-in-a", partition 0 + And I have a stream with name "purge-stream-b" + And I have a topic with name "topic-in-b" in stream "purge-stream-b" with 1 partition + And I have sent 10 messages to stream "purge-stream-b", topic "topic-in-b", partition 0 + When I purge the stream "purge-stream-a" + And I poll messages from stream "purge-stream-a", topic "topic-in-a", partition 0 starting from offset 0 + Then I should receive 0 messages + And I poll messages from stream "purge-stream-b", topic "topic-in-b", partition 0 starting from offset 0 + Then I should receive 10 messages + + Scenario: Purge topic does not affect other topic + Given I have a stream with name "multi-topic-stream" + And I have a topic with name "topic-to-purge" in stream "multi-topic-stream" with 1 partition + And I have a topic with name "topic-to-keep" in stream "multi-topic-stream" with 1 partition + And I have sent 10 messages to stream "multi-topic-stream", topic "topic-to-purge", partition 0 + And I have sent 10 messages to stream "multi-topic-stream", topic "topic-to-keep", partition 0 + When I purge topic "topic-to-purge" in stream "multi-topic-stream" + And I poll messages from stream "multi-topic-stream", topic "topic-to-purge", partition 0 starting from offset 0 + Then I should receive 0 messages + And I poll messages from stream "multi-topic-stream", topic "topic-to-keep", partition 0 starting from offset 0 + Then I should receive 10 messages + + Scenario: Purge non-existing stream should return an error + When I try to purge a non-existing stream + Then the purge operation should fail with a stream not found error + + Scenario: Purge non-existing topic should return an error + Given I have a stream with name "stream-for-purge" + When I try to purge a non-existing topic in stream "stream-for-purge" + Then the purge topic operation should fail with a topic not found error From 25478a75660fe0671681af9d8b8e5715de6b86b0 Mon Sep 17 00:00:00 2001 From: chengxi Date: Sat, 4 Jul 2026 14:32:57 -0400 Subject: [PATCH 3/6] test(sdk): update the ./scripts/run-bdd-tests.sh for new scenario --- scripts/run-bdd-tests.sh | 33 +++++++++++++++++++++------------ 1 file changed, 21 insertions(+), 12 deletions(-) diff --git a/scripts/run-bdd-tests.sh b/scripts/run-bdd-tests.sh index 148a564b3a..cb19cdca4c 100755 --- a/scripts/run-bdd-tests.sh +++ b/scripts/run-bdd-tests.sh @@ -36,7 +36,7 @@ usage(){ log "Usage: $0 [--coverage] [feature]" log "" log " sdk: rust | python | php | go | go-race | node | csharp | java | cpp | all | clean (default: all)" - log " feature: basic_messaging | leader_redirection | all (default: all)" + log " feature: basic_messaging | leader_redirection | stream_topic_purge | all (default: all)" log "" log "Examples:" log " $0 rust # run all features for Rust" @@ -46,7 +46,7 @@ usage(){ } case "$FEATURE" in - basic_messaging|leader_redirection|all) ;; + basic_messaging|leader_redirection|stream_topic_purge|all) ;; *) log "Unknown feature: ${FEATURE}" usage @@ -66,7 +66,7 @@ ALL_COMPOSE_FILES=( COMPOSE_FILES=(-f docker-compose.yml) case "$FEATURE" in - basic_messaging|leader_redirection|all) + basic_messaging|leader_redirection|stream_topic_purge|all) COMPOSE_FILES+=(-f docker-compose.server.yml) ;; esac case "$FEATURE" in @@ -90,20 +90,29 @@ if [ "$COVERAGE" = "1" ]; then log "📊 Coverage collection enabled → reports will be in ./reports/" fi +unsupported() { + local feature="$1" svc="$2" + if [ "$SDK" = "all" ]; then + log "⚠️ skipping ${svc%-bdd} (does not support ${feature})" + return 0 + else + log "❌ ${SDK} does not support feature '${feature}'" + return 1 + fi +} + run_suite(){ local svc="$1" emoji="$2" label="$3" if [ "$FEATURE" = "leader_redirection" ]; then case "$svc" in rust-bdd|go-bdd|csharp-bdd) ;; - *) - if [ "$SDK" = "all" ]; then - log "⚠️ skipping ${svc%-bdd} (does not support ${FEATURE})" - return 0 - else - log "❌ ${SDK} does not support feature '${FEATURE}'" - return 1 - fi - ;; + *) unsupported "$FEATURE" "$svc" || return 1; return 0 ;; + esac + fi + if [ "$FEATURE" = "stream_topic_purge" ]; then + case "$svc" in + rust-bdd|go-bdd) ;; + *) unsupported "$FEATURE" "$svc" || return 1; return 0 ;; esac fi From 737b29f99db8ca20a2fdbd00eadec3df305ae9d6 Mon Sep 17 00:00:00 2001 From: chengxi Date: Sat, 4 Jul 2026 13:57:15 -0400 Subject: [PATCH 4/6] test(go): implement stream_topic_purge scenario for go --- bdd/go/tests/basic_messaging.go | 45 +----- bdd/go/tests/helper.go | 129 ++++++++++++++++ bdd/go/tests/leader_redirection.go | 25 +-- bdd/go/tests/stream_topic_purge.go | 239 +++++++++++++++++++++++++++++ 4 files changed, 378 insertions(+), 60 deletions(-) create mode 100644 bdd/go/tests/helper.go create mode 100644 bdd/go/tests/stream_topic_purge.go diff --git a/bdd/go/tests/basic_messaging.go b/bdd/go/tests/basic_messaging.go index ceb4137afc..0aafcff620 100644 --- a/bdd/go/tests/basic_messaging.go +++ b/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, diff --git a/bdd/go/tests/helper.go b/bdd/go/tests/helper.go new file mode 100644 index 0000000000..2353d6da35 --- /dev/null +++ b/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 +} diff --git a/bdd/go/tests/leader_redirection.go b/bdd/go/tests/leader_redirection.go index 9ee79d3a83..c95064aa60 100644 --- a/bdd/go/tests/leader_redirection.go +++ b/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 } diff --git a/bdd/go/tests/stream_topic_purge.go b/bdd/go/tests/stream_topic_purge.go new file mode 100644 index 0000000000..3416574746 --- /dev/null +++ b/bdd/go/tests/stream_topic_purge.go @@ -0,0 +1,239 @@ +// 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" + "errors" + "fmt" + "math" + + iggcon "github.com/apache/iggy/foreign/go/contracts" + ierror "github.com/apache/iggy/foreign/go/errors" + "github.com/cucumber/godog" +) + +type purgeCtxKey struct{} + +type purgeCtx struct { + client iggcon.Client + streamIDs map[string]uint32 + topicIDs map[string]uint32 + lastErr error + lastPollMessages *iggcon.PolledMessage +} + +func getPurgeCtx(ctx context.Context) *purgeCtx { + return ctx.Value(purgeCtxKey{}).(*purgeCtx) +} + +type purgeSteps struct{} + +func (s purgeSteps) givenRunningServer(ctx context.Context) error { + c := getPurgeCtx(ctx) + cli, err := connectToServer(ctx, defaultServerAddress()) + if err != nil { + return err + } + c.client = cli + return nil +} + +func (s purgeSteps) givenAuthenticatedAsRoot(ctx context.Context) error { + return loginAsRoot(ctx, getPurgeCtx(ctx).client) +} + +func (s purgeSteps) givenHaveStream(ctx context.Context, name string) error { + c := getPurgeCtx(ctx) + stream, err := c.client.CreateStream(ctx, name) + if err != nil { + return fmt.Errorf("error creating stream: %w", err) + } + c.streamIDs[name] = stream.Id + return nil +} + +func (s purgeSteps) givenHaveTopic(ctx context.Context, topicName, streamName string, partitionsCount uint32) error { + c := getPurgeCtx(ctx) + streamID, ok := c.streamIDs[streamName] + if !ok { + return fmt.Errorf("stream %q not found", streamName) + } + streamIdentifier, _ := iggcon.NewIdentifier(streamID) + topic, err := c.client.CreateTopic( + ctx, + streamIdentifier, + topicName, + partitionsCount, + iggcon.CompressionAlgorithmNone, + iggcon.IggyExpiryNeverExpire, + 0, + nil, + ) + if err != nil { + return fmt.Errorf("error creating topic: %w", err) + } + c.topicIDs[topicName] = topic.Id + return nil +} + +func (s purgeSteps) givenHaveSentMessages(ctx context.Context, messagesCount uint32, streamName, topicName string, partitionID uint32) error { + c := getPurgeCtx(ctx) + streamID, ok := c.streamIDs[streamName] + if !ok { + return fmt.Errorf("stream %q not found", streamName) + } + topicID, ok := c.topicIDs[topicName] + if !ok { + return fmt.Errorf("topic %q not found", topicName) + } + streamIdentifier, _ := iggcon.NewIdentifier(streamID) + topicIdentifier, _ := iggcon.NewIdentifier(topicID) + if _, err := sendTestMessages(ctx, c.client, streamIdentifier, topicIdentifier, partitionID, messagesCount); err != nil { + return err + } + return nil +} + +func (s purgeSteps) whenPurgeStream(ctx context.Context, streamName string) error { + c := getPurgeCtx(ctx) + streamID, ok := c.streamIDs[streamName] + if !ok { + return fmt.Errorf("stream %q not found", streamName) + } + streamIdentifier, _ := iggcon.NewIdentifier(streamID) + if err := c.client.PurgeStream(ctx, streamIdentifier); err != nil { + return fmt.Errorf("error purging stream: %w", err) + } + return nil +} + +func (s purgeSteps) whenPurgeTopic(ctx context.Context, topicName, streamName string) error { + c := getPurgeCtx(ctx) + streamID, ok := c.streamIDs[streamName] + if !ok { + return fmt.Errorf("stream %q not found", streamName) + } + topicID, ok := c.topicIDs[topicName] + if !ok { + return fmt.Errorf("topic %q not found", topicName) + } + streamIdentifier, _ := iggcon.NewIdentifier(streamID) + topicIdentifier, _ := iggcon.NewIdentifier(topicID) + if err := c.client.PurgeTopic(ctx, streamIdentifier, topicIdentifier); err != nil { + return fmt.Errorf("error purging topic: %w", err) + } + return nil +} + +func (s purgeSteps) whenTryPurgeNonExistingStream(ctx context.Context) error { + c := getPurgeCtx(ctx) + streamIdentifier, _ := iggcon.NewIdentifier[uint32](math.MaxUint32) + c.lastErr = c.client.PurgeStream(ctx, streamIdentifier) + return nil +} + +func (s purgeSteps) whenTryPurgeNonExistingTopic(ctx context.Context, streamName string) error { + c := getPurgeCtx(ctx) + streamID, ok := c.streamIDs[streamName] + if !ok { + return fmt.Errorf("stream %q not found", streamName) + } + streamIdentifier, _ := iggcon.NewIdentifier(streamID) + topicIdentifier, _ := iggcon.NewIdentifier[uint32](math.MaxUint32) + c.lastErr = c.client.PurgeTopic(ctx, streamIdentifier, topicIdentifier) + return nil +} + +func (s purgeSteps) whenPollMessages(ctx context.Context, streamName, topicName string, partitionID uint32, startOffset uint64) error { + c := getPurgeCtx(ctx) + streamID, ok := c.streamIDs[streamName] + if !ok { + return fmt.Errorf("stream %q not found", streamName) + } + topicID, ok := c.topicIDs[topicName] + if !ok { + return fmt.Errorf("topic %q not found", topicName) + } + streamIdentifier, _ := iggcon.NewIdentifier(streamID) + topicIdentifier, _ := iggcon.NewIdentifier(topicID) + polled, err := pollTestMessages(ctx, c.client, streamIdentifier, topicIdentifier, partitionID, startOffset, 100) + if err != nil { + return err + } + c.lastPollMessages = polled + return nil +} + +func (s purgeSteps) thenShouldReceiveMessages(ctx context.Context, expectedCount uint32) error { + return assertMessageCount(getPurgeCtx(ctx).lastPollMessages, expectedCount) +} + +func (s purgeSteps) thenPurgeFailsWithStreamNotFound(ctx context.Context) error { + c := getPurgeCtx(ctx) + if c.lastErr == nil { + return errors.New("expected an error when purging non-existing stream, got nil") + } + if !errors.Is(c.lastErr, ierror.ErrStreamIdNotFound) { + return fmt.Errorf("expected ErrStreamIdNotFound, got: %v", c.lastErr) + } + return nil +} + +func (s purgeSteps) thenPurgeTopicFailsWithTopicNotFound(ctx context.Context) error { + c := getPurgeCtx(ctx) + if c.lastErr == nil { + return errors.New("expected an error when purging non-existing topic, got nil") + } + if !errors.Is(c.lastErr, ierror.ErrTopicIdNotFound) { + return fmt.Errorf("expected ErrTopicIdNotFound, got: %v", c.lastErr) + } + return nil +} + +func initPurgeScenario(sc *godog.ScenarioContext) { + sc.Before(func(ctx context.Context, _ *godog.Scenario) (context.Context, error) { + return context.WithValue(context.Background(), purgeCtxKey{}, &purgeCtx{ + streamIDs: make(map[string]uint32), + topicIDs: make(map[string]uint32), + }), nil + }) + s := &purgeSteps{} + sc.Step(`^I have a running Iggy server$`, s.givenRunningServer) + sc.Step(`^I am authenticated as the root user$`, s.givenAuthenticatedAsRoot) + sc.Step(`^I have a stream with name "([^"]*)"$`, s.givenHaveStream) + sc.Step(`^I have a topic with name "([^"]*)" in stream "([^"]*)" with (\d+) partition[s]?$`, s.givenHaveTopic) + sc.Step(`^I have sent (\d+) messages to stream "([^"]*)", topic "([^"]*)", partition (\d+)$`, s.givenHaveSentMessages) + sc.Step(`^I purge the stream "([^"]*)"$`, s.whenPurgeStream) + sc.Step(`^I purge topic "([^"]*)" in stream "([^"]*)"$`, s.whenPurgeTopic) + sc.Step(`^I try to purge a non-existing stream$`, s.whenTryPurgeNonExistingStream) + sc.Step(`^I try to purge a non-existing topic in stream "([^"]*)"$`, s.whenTryPurgeNonExistingTopic) + sc.Step(`^I poll messages from stream "([^"]*)", topic "([^"]*)", partition (\d+) starting from offset (\d+)$`, s.whenPollMessages) + sc.Step(`^I should receive (\d+) messages$`, s.thenShouldReceiveMessages) + sc.Step(`^the purge operation should fail with a stream not found error$`, s.thenPurgeFailsWithStreamNotFound) + sc.Step(`^the purge topic operation should fail with a topic not found error$`, s.thenPurgeTopicFailsWithTopicNotFound) + sc.After(func(ctx context.Context, _ *godog.Scenario, scErr error) (context.Context, error) { + c := getPurgeCtx(ctx) + if c.client != nil { + if err := c.client.Close(); err != nil { + scErr = errors.Join(scErr, fmt.Errorf("error closing client: %w", err)) + } + } + return ctx, scErr + }) +} From 78f08f1d969152b147e3b940038b04de438fb278 Mon Sep 17 00:00:00 2001 From: chengxi Date: Sat, 4 Jul 2026 17:15:51 -0400 Subject: [PATCH 5/6] fix(server): clear journal and in-flight data on partition purge --- core/server/src/shard/system/segments.rs | 4 +++- core/server/src/streaming/partitions/journal.rs | 7 +++++++ 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/core/server/src/shard/system/segments.rs b/core/server/src/shard/system/segments.rs index fb48846a76..e5248969a5 100644 --- a/core/server/src/shard/system/segments.rs +++ b/core/server/src/shard/system/segments.rs @@ -17,6 +17,7 @@ use crate::configs::cache_indexes::CacheIndexesConfig; use crate::shard::IggyShard; +use crate::streaming::partitions::journal::Journal; use crate::streaming::segments::Segment; use iggy_common::{ConsumerKind, IggyError, IggyExpiry, IggyTimestamp, MaxTopicSize}; use server_common::sharding::IggyNamespace; @@ -508,8 +509,9 @@ impl IggyShard { let mut partitions = self.local_partitions.borrow_mut(); if let Some(partition) = partitions.get_mut(namespace) { + partition.log.journal_mut().reset(); + partition.log.clear_in_flight(); partition.log.add_persisted_segment(segment, storage); - // Reset offset when starting fresh with a new segment at offset 0 partition .offset .store(start_offset, std::sync::atomic::Ordering::SeqCst); diff --git a/core/server/src/streaming/partitions/journal.rs b/core/server/src/streaming/partitions/journal.rs index 87bd9cfcb5..c4384e10a0 100644 --- a/core/server/src/streaming/partitions/journal.rs +++ b/core/server/src/streaming/partitions/journal.rs @@ -137,6 +137,11 @@ impl Journal for MemoryMessageJournal { self.batches.is_empty() } + fn reset(&mut self) { + self.batches = IggyMessagesBatchSet::default(); + self.inner = Inner::default(); + } + fn inner(&self) -> &Self::Inner { &self.inner } @@ -204,6 +209,8 @@ pub trait Journal { /// Timestamp of last message in journal, or None if empty. fn last_timestamp(&self) -> Option; + fn reset(&mut self); + // `flush` is only useful in case of an journal that has disk backed WAL. // This could be merged together with `append`, but not doing this for two reasons. // 1. In case of the `Journal` being used as part of structure that utilizes interior mutability, async with borrow_mut is not possible. From f5d74d92013689a79ca24fd522265751ffe3ffa2 Mon Sep 17 00:00:00 2001 From: chengxi Date: Sat, 4 Jul 2026 17:40:31 -0400 Subject: [PATCH 6/6] test(rust): implement stream_topic_purge bdd test for rust --- bdd/docker-compose.yml | 2 + bdd/rust/Cargo.toml | 5 + bdd/rust/tests/common/mod.rs | 1 + bdd/rust/tests/common/purge_context.rs | 31 +++ bdd/rust/tests/steps/mod.rs | 1 + bdd/rust/tests/steps/purge.rs | 284 +++++++++++++++++++++++++ bdd/rust/tests/stream_topic_purge.rs | 28 +++ 7 files changed, 352 insertions(+) create mode 100644 bdd/rust/tests/common/purge_context.rs create mode 100644 bdd/rust/tests/steps/purge.rs create mode 100644 bdd/rust/tests/stream_topic_purge.rs diff --git a/bdd/docker-compose.yml b/bdd/docker-compose.yml index 0c6076634d..b08cac3f82 100644 --- a/bdd/docker-compose.yml +++ b/bdd/docker-compose.yml @@ -30,6 +30,7 @@ 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 @@ -37,6 +38,7 @@ services: 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: diff --git a/bdd/rust/Cargo.toml b/bdd/rust/Cargo.toml index ebedb19d28..5ff683be31 100644 --- a/bdd/rust/Cargo.toml +++ b/bdd/rust/Cargo.toml @@ -44,3 +44,8 @@ required-features = ["bdd"] name = "leader_redirection" harness = false required-features = ["bdd"] + +[[test]] +name = "stream_topic_purge" +harness = false +required-features = ["bdd"] diff --git a/bdd/rust/tests/common/mod.rs b/bdd/rust/tests/common/mod.rs index 1a0e223db5..a1622fbf09 100644 --- a/bdd/rust/tests/common/mod.rs +++ b/bdd/rust/tests/common/mod.rs @@ -17,3 +17,4 @@ pub mod global_context; pub mod leader_context; +pub mod purge_context; diff --git a/bdd/rust/tests/common/purge_context.rs b/bdd/rust/tests/common/purge_context.rs new file mode 100644 index 0000000000..9edfdb755d --- /dev/null +++ b/bdd/rust/tests/common/purge_context.rs @@ -0,0 +1,31 @@ +// 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. + +use cucumber::World; +use iggy::clients::client::IggyClient; +use iggy::prelude::{IggyError, PolledMessages}; +use std::collections::HashMap; + +#[derive(Debug, World, Default)] +pub struct PurgeContext { + pub client: Option, + pub server_addr: Option, + pub stream_ids: HashMap, + pub topic_ids: HashMap, + pub last_polled_messages: Option, + pub last_error: Option, +} diff --git a/bdd/rust/tests/steps/mod.rs b/bdd/rust/tests/steps/mod.rs index 907084ec3a..42d33c0bad 100644 --- a/bdd/rust/tests/steps/mod.rs +++ b/bdd/rust/tests/steps/mod.rs @@ -18,6 +18,7 @@ pub mod auth; pub mod leader_redirection; pub mod messages; +pub mod purge; pub mod server; pub mod streams; pub mod topics; diff --git a/bdd/rust/tests/steps/purge.rs b/bdd/rust/tests/steps/purge.rs new file mode 100644 index 0000000000..09e7cd31a0 --- /dev/null +++ b/bdd/rust/tests/steps/purge.rs @@ -0,0 +1,284 @@ +// 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. + +use crate::common::purge_context::PurgeContext; +use crate::helpers::test_data::create_test_messages; +use cucumber::{given, then, when}; +use iggy::prelude::{ + Client, ClientWrapper, CompressionAlgorithm, Consumer, ConsumerKind, DEFAULT_ROOT_PASSWORD, + DEFAULT_ROOT_USERNAME, Identifier, IggyError, IggyExpiry, MaxTopicSize, MessageClient, + Partitioning, PollingStrategy, StreamClient, SystemClient, TcpClient, TcpClientConfig, + TopicClient, UserClient, +}; +use std::sync::Arc; + +#[given("I have a running Iggy server")] +pub async fn given_running_server(world: &mut PurgeContext) { + let server_addr = + std::env::var("IGGY_TCP_ADDRESS").unwrap_or_else(|_| "localhost:8090".to_string()); + world.server_addr = Some(server_addr); +} + +#[given("I am authenticated as the root user")] +pub async fn given_authenticated_as_root(world: &mut PurgeContext) { + let server_addr = world + .server_addr + .as_ref() + .expect("Server should be running") + .clone(); + + let config = TcpClientConfig { + server_address: server_addr, + ..TcpClientConfig::default() + }; + + let tcp_client = TcpClient::create(Arc::new(config)).expect("Failed to create TCP client"); + Client::connect(&tcp_client) + .await + .expect("Client should connect"); + + let client = + iggy::clients::client::IggyClient::create(ClientWrapper::Tcp(tcp_client), None, None); + + client.ping().await.expect("Server should respond to ping"); + client + .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD) + .await + .expect("Failed to login as root"); + + world.client = Some(client); +} + +#[given(regex = r#"^I have a stream with name "([^"]*)"$"#)] +pub async fn given_have_stream(world: &mut PurgeContext, name: String) { + let client = world.client.as_ref().expect("Client should be available"); + let stream = client + .create_stream(&name) + .await + .expect("Should be able to create stream"); + world.stream_ids.insert(name, stream.id); +} + +#[given( + regex = r#"^I have a topic with name "([^"]*)" in stream "([^"]*)" with (\d+) partition[s]?$"# +)] +pub async fn given_have_topic( + world: &mut PurgeContext, + topic_name: String, + stream_name: String, + partitions_count: u32, +) { + let client = world.client.as_ref().expect("Client should be available"); + let stream_id = world + .stream_ids + .get(&stream_name) + .copied() + .unwrap_or_else(|| panic!("Stream {stream_name:?} not found")); + + let topic = client + .create_topic( + &Identifier::numeric(stream_id).unwrap(), + &topic_name, + partitions_count, + CompressionAlgorithm::default(), + None, + IggyExpiry::NeverExpire, + MaxTopicSize::ServerDefault, + ) + .await + .expect("Should be able to create topic"); + world.topic_ids.insert(topic_name, topic.id); +} + +#[given( + regex = r#"^I have sent (\d+) messages to stream "([^"]*)", topic "([^"]*)", partition (\d+)$"# +)] +pub async fn given_have_sent_messages( + world: &mut PurgeContext, + messages_count: u32, + stream_name: String, + topic_name: String, + partition_id: u32, +) { + let client = world.client.as_ref().expect("Client should be available"); + let stream_id = world + .stream_ids + .get(&stream_name) + .copied() + .unwrap_or_else(|| panic!("Stream {stream_name:?} not found")); + let topic_id = world + .topic_ids + .get(&topic_name) + .copied() + .unwrap_or_else(|| panic!("Topic {topic_name:?} not found")); + + let mut messages = create_test_messages(messages_count); + client + .send_messages( + &Identifier::numeric(stream_id).unwrap(), + &Identifier::numeric(topic_id).unwrap(), + &Partitioning::partition_id(partition_id), + &mut messages, + ) + .await + .expect("Should be able to send messages"); +} + +#[when(regex = r#"^I purge the stream "([^"]*)"$"#)] +pub async fn when_purge_stream(world: &mut PurgeContext, stream_name: String) { + let client = world.client.as_ref().expect("Client should be available"); + let stream_id = world + .stream_ids + .get(&stream_name) + .copied() + .unwrap_or_else(|| panic!("Stream {stream_name:?} not found")); + + client + .purge_stream(&Identifier::numeric(stream_id).unwrap()) + .await + .expect("Should be able to purge stream"); +} + +#[when(regex = r#"^I purge topic "([^"]*)" in stream "([^"]*)"$"#)] +pub async fn when_purge_topic(world: &mut PurgeContext, topic_name: String, stream_name: String) { + let client = world.client.as_ref().expect("Client should be available"); + let stream_id = world + .stream_ids + .get(&stream_name) + .copied() + .unwrap_or_else(|| panic!("Stream {stream_name:?} not found")); + let topic_id = world + .topic_ids + .get(&topic_name) + .copied() + .unwrap_or_else(|| panic!("Topic {topic_name:?} not found")); + + client + .purge_topic( + &Identifier::numeric(stream_id).unwrap(), + &Identifier::numeric(topic_id).unwrap(), + ) + .await + .expect("Should be able to purge topic"); +} + +#[when("I try to purge a non-existing stream")] +pub async fn when_try_purge_non_existing_stream(world: &mut PurgeContext) { + let client = world.client.as_ref().expect("Client should be available"); + let result = client + .purge_stream(&Identifier::numeric(u32::MAX).unwrap()) + .await; + world.last_error = result.err(); +} + +#[when(regex = r#"^I try to purge a non-existing topic in stream "([^"]*)"$"#)] +pub async fn when_try_purge_non_existing_topic(world: &mut PurgeContext, stream_name: String) { + let client = world.client.as_ref().expect("Client should be available"); + let stream_id = world + .stream_ids + .get(&stream_name) + .copied() + .unwrap_or_else(|| panic!("Stream {stream_name:?} not found")); + + let result = client + .purge_topic( + &Identifier::numeric(stream_id).unwrap(), + &Identifier::numeric(u32::MAX).unwrap(), + ) + .await; + world.last_error = result.err(); +} + +#[when( + regex = r#"^I poll messages from stream "([^"]*)", topic "([^"]*)", partition (\d+) starting from offset (\d+)$"# +)] +pub async fn when_poll_messages( + world: &mut PurgeContext, + stream_name: String, + topic_name: String, + partition_id: u32, + start_offset: u64, +) { + let client = world.client.as_ref().expect("Client should be available"); + let stream_id = world + .stream_ids + .get(&stream_name) + .copied() + .unwrap_or_else(|| panic!("Stream {stream_name:?} not found")); + let topic_id = world + .topic_ids + .get(&topic_name) + .copied() + .unwrap_or_else(|| panic!("Topic {topic_name:?} not found")); + + let consumer = Consumer { + kind: ConsumerKind::Consumer, + id: Identifier::numeric(1).unwrap(), + }; + + let polled_messages = client + .poll_messages( + &Identifier::numeric(stream_id).unwrap(), + &Identifier::numeric(topic_id).unwrap(), + Some(partition_id), + &consumer, + &PollingStrategy::offset(start_offset), + 100, + false, + ) + .await + .expect("Should be able to poll messages"); + + world.last_polled_messages = Some(polled_messages); +} + +#[then(regex = r"^I should receive (\d+) messages$")] +pub async fn then_should_receive_messages(world: &mut PurgeContext, expected_count: u32) { + let polled_messages = world + .last_polled_messages + .as_ref() + .expect("Should have polled messages"); + assert_eq!( + polled_messages.messages.len() as u32, + expected_count, + "Should receive exactly {expected_count} messages", + ); +} + +#[then("the purge operation should fail with a stream not found error")] +pub async fn then_purge_fails_with_stream_not_found(world: &mut PurgeContext) { + let err = world + .last_error + .as_ref() + .expect("Should have captured an error"); + assert!( + matches!(err, IggyError::StreamIdNotFound(_)), + "Expected StreamIdNotFound, got: {err:?}", + ); +} + +#[then("the purge topic operation should fail with a topic not found error")] +pub async fn then_purge_topic_fails_with_topic_not_found(world: &mut PurgeContext) { + let err = world + .last_error + .as_ref() + .expect("Should have captured an error"); + assert!( + matches!(err, IggyError::TopicIdNotFound(_, _)), + "Expected TopicIdNotFound, got: {err:?}", + ); +} diff --git a/bdd/rust/tests/stream_topic_purge.rs b/bdd/rust/tests/stream_topic_purge.rs new file mode 100644 index 0000000000..a28e45837a --- /dev/null +++ b/bdd/rust/tests/stream_topic_purge.rs @@ -0,0 +1,28 @@ +// 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. + +pub(crate) mod common; +pub(crate) mod helpers; +pub(crate) mod steps; + +use crate::common::purge_context::PurgeContext; +use cucumber::World; + +#[tokio::main] +async fn main() { + PurgeContext::run("../../bdd/scenarios/stream_topic_purge.feature").await; +}