diff --git a/weed/shell/command_mq_topic_truncate.go b/weed/shell/command_mq_topic_truncate.go new file mode 100644 index 000000000..1128e3b88 --- /dev/null +++ b/weed/shell/command_mq_topic_truncate.go @@ -0,0 +1,180 @@ +package shell + +import ( + "context" + "flag" + "fmt" + "io" + "strings" + + "github.com/seaweedfs/seaweedfs/weed/mq/topic" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +func init() { + Commands = append(Commands, &commandMqTopicTruncate{}) +} + +type commandMqTopicTruncate struct { +} + +func (c *commandMqTopicTruncate) Name() string { + return "mq.topic.truncate" +} + +func (c *commandMqTopicTruncate) Help() string { + return `clear all data from a topic while preserving topic structure + + Example: + mq.topic.truncate -namespace -topic + + This command removes all log files and parquet files from all partitions + of the specified topic, while keeping the topic configuration intact. +` +} + +func (c *commandMqTopicTruncate) HasTag(CommandTag) bool { + return false +} + +func (c *commandMqTopicTruncate) Do(args []string, commandEnv *CommandEnv, writer io.Writer) error { + // parse parameters + mqCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError) + namespace := mqCommand.String("namespace", "", "namespace name") + topicName := mqCommand.String("topic", "", "topic name") + if err := mqCommand.Parse(args); err != nil { + return err + } + + if *namespace == "" { + return fmt.Errorf("namespace is required") + } + if *topicName == "" { + return fmt.Errorf("topic name is required") + } + + // Verify topic exists by trying to read its configuration + t := topic.NewTopic(*namespace, *topicName) + + err := commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + _, err := t.ReadConfFile(client) + if err != nil { + return fmt.Errorf("topic %s.%s does not exist or cannot be read: %v", *namespace, *topicName, err) + } + return nil + }) + if err != nil { + return err + } + + fmt.Fprintf(writer, "Truncating topic %s.%s...\n", *namespace, *topicName) + + // Discover and clear all partitions + partitions, err := c.discoverTopicPartitions(commandEnv, t) + if err != nil { + return fmt.Errorf("failed to discover topic partitions: %v", err) + } + + if len(partitions) == 0 { + fmt.Fprintf(writer, "No partitions found for topic %s.%s\n", *namespace, *topicName) + return nil + } + + fmt.Fprintf(writer, "Found %d partitions, clearing data...\n", len(partitions)) + + // Clear data from each partition + totalFilesDeleted := 0 + for _, partitionPath := range partitions { + filesDeleted, err := c.clearPartitionData(commandEnv, partitionPath, writer) + if err != nil { + fmt.Fprintf(writer, "Warning: failed to clear partition %s: %v\n", partitionPath, err) + continue + } + totalFilesDeleted += filesDeleted + fmt.Fprintf(writer, "Cleared partition: %s (%d files)\n", partitionPath, filesDeleted) + } + + fmt.Fprintf(writer, "Successfully truncated topic %s.%s - deleted %d files from %d partitions\n", + *namespace, *topicName, totalFilesDeleted, len(partitions)) + + return nil +} + +// discoverTopicPartitions discovers all partition directories for a topic +func (c *commandMqTopicTruncate) discoverTopicPartitions(commandEnv *CommandEnv, t topic.Topic) ([]string, error) { + var partitionPaths []string + + // Scan the topic directory for version directories (e.g., v2025-09-01-07-16-34) + err := filer_pb.ReadDirAllEntries(context.Background(), commandEnv, util.FullPath(t.Dir()), "", func(versionEntry *filer_pb.Entry, isLast bool) error { + if !versionEntry.IsDirectory { + return nil // Skip non-directories + } + + // Parse version timestamp from directory name (e.g., "v2025-09-01-07-16-34") + _, parseErr := topic.ParseTopicVersion(versionEntry.Name) + if parseErr != nil { + // Skip directories that don't match the version format + return nil + } + + // Scan partition directories within this version (e.g., 0000-0630) + versionDir := fmt.Sprintf("%s/%s", t.Dir(), versionEntry.Name) + return filer_pb.ReadDirAllEntries(context.Background(), commandEnv, util.FullPath(versionDir), "", func(partitionEntry *filer_pb.Entry, isLast bool) error { + if !partitionEntry.IsDirectory { + return nil // Skip non-directories + } + + // Parse partition boundary from directory name (e.g., "0000-0630") + rangeStart, rangeStop := topic.ParsePartitionBoundary(partitionEntry.Name) + if rangeStart == rangeStop { + return nil // Skip invalid partition names + } + + // Add this partition path to the list + partitionPath := fmt.Sprintf("%s/%s", versionDir, partitionEntry.Name) + partitionPaths = append(partitionPaths, partitionPath) + return nil + }) + }) + + return partitionPaths, err +} + +// clearPartitionData deletes all data files (log files, parquet files) from a partition directory +// Returns the number of files deleted +func (c *commandMqTopicTruncate) clearPartitionData(commandEnv *CommandEnv, partitionPath string, writer io.Writer) (int, error) { + filesDeleted := 0 + + err := filer_pb.ReadDirAllEntries(context.Background(), commandEnv, util.FullPath(partitionPath), "", func(entry *filer_pb.Entry, isLast bool) error { + if entry.IsDirectory { + return nil // Skip subdirectories + } + + fileName := entry.Name + + // Preserve configuration files + if strings.HasSuffix(fileName, ".conf") || + strings.HasSuffix(fileName, ".config") || + fileName == "topic.conf" || + fileName == "partition.conf" { + fmt.Fprintf(writer, " Preserving config file: %s\n", fileName) + return nil + } + + // Delete all data files (log files, parquet files, offset files, etc.) + deleteErr := filer_pb.Remove(context.Background(), commandEnv, partitionPath, fileName, false, true, true, false, nil) + + if deleteErr != nil { + fmt.Fprintf(writer, " Warning: failed to delete %s/%s: %v\n", partitionPath, fileName, deleteErr) + // Continue with other files rather than failing entirely + } else { + fmt.Fprintf(writer, " Deleted: %s\n", fileName) + filesDeleted++ + } + + return nil + }) + + return filesDeleted, err +}