ongoing

package
v1.1.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Mar 13, 2024 License: MIT Imports: 11 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source
var Cmd = &cobra.Command{
	Use:   "ongoing",
	Short: "List, cancel, or wait for ongoing reassignments",
	RunE: func(cmd *cobra.Command, args []string) error {
		client := &kafka.Client{Addr: kafka.TCP(vars.BootstrapServer)}

		if cancel {
			ongoing, err := listOngoing(client)
			if err != nil {
				return fmt.Errorf("problem listing ongoing reassignments: %w", err)
			}

			err = cancelReassignments(client, ongoing)
			if err != nil {
				return fmt.Errorf("problem canceling partition reassignments: %w", err)
			}

			fmt.Println("Ongoing reassignments cancelled:")
			fmt.Println()
			printOngoing(ongoing)
			return nil
		}

		if wait {
			var totalWait time.Duration
			err := Wait(client, func(err error, wait time.Duration) {
				totalWait += wait
				fmt.Printf("Waited %s for ongoing reassignments to complete...\n", totalWait.Truncate(time.Second).String())
			})
			if err != nil {
				return fmt.Errorf("problem waiting for reassignments: %w", err)
			}

			fmt.Println("No ongoing reassignments.")
			return nil
		}

		ongoing, err := listOngoing(client)
		if err != nil {
			return fmt.Errorf("problem listing ongoing reassignments: %w", err)
		}
		printOngoing(ongoing)

		return nil
	},
}

Functions

func Wait

func Wait(client *kafka.Client, notify func(error, time.Duration)) error

Types

This section is empty.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL