[CF-4208] Add --page-size to on-prem Flink list commands

internal/flink/command.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"math"
 	"net/http"
 
 	"github.com/spf13/cobra"
@@ -125,6 +126,21 @@ func addCmfFlagSet(cmd *cobra.Command) {
 	cmd.Flags().String("certificate-authority-path", "", `Path to a PEM-encoded Certificate Authority to verify the Confluent Manager for Apache Flink connection. Environment variable "CONFLUENT_CMF_CERTIFICATE_AUTHORITY_PATH" may be set in place of this flag.`)
 }
 
+func addPageSizeFlag(cmd *cobra.Command) {
+	cmd.Flags().Int("page-size", 0, "Number of results to fetch per API request. Defaults to 100.")
+}
+
+func getPageSize(cmd *cobra.Command) (int32, error) {
+	pageSize, err := cmd.Flags().GetInt("page-size")
+	if err != nil {
+		return 0, err
+	}
+	if pageSize < 0 || pageSize > math.MaxInt32 {
+		return 0, fmt.Errorf("`--page-size` must be between 0 and %d", math.MaxInt32)
+	}
+	return int32(pageSize), nil
+}
+
 func (c *command) createContext() context.Context {
 	if !c.Config.IsOnPremLogin() {
 		return context.Background()

internal/flink/command_application_event_list.go

@@ -17,6 +17,7 @@ func (c *command) newApplicationEventListCommand() *cobra.Command {
 
 	cmd.Flags().String("environment", "", "Name of the Flink environment.")
 	cmd.Flags().String("application", "", "Name of the Flink application.")
+	addPageSizeFlag(cmd)
 	addCmfFlagSet(cmd)
 	pcmd.AddOutputFlag(cmd)
 
@@ -37,12 +38,17 @@ func (c *command) applicationEventList(cmd *cobra.Command, _ []string) error {
 		return err
 	}
 
+	pageSize, err := getPageSize(cmd)
+	if err != nil {
+		return err
+	}
+
 	client, err := c.GetCmfClient(cmd)
 	if err != nil {
 		return err
 	}
 
-	events, err := client.ListApplicationEvents(c.createContext(), environment, application)
+	events, err := client.ListApplicationEvents(c.createContext(), environment, application, pageSize)
 	if err != nil {
 		return err
 	}

internal/flink/command_application_instance_list.go

@@ -17,6 +17,7 @@ func (c *command) newApplicationInstanceListCommand() *cobra.Command {
 
 	cmd.Flags().String("environment", "", "Name of the Flink environment.")
 	cmd.Flags().String("application", "", "Name of the Flink application.")
+	addPageSizeFlag(cmd)
 	addCmfFlagSet(cmd)
 	pcmd.AddOutputFlag(cmd)
 
@@ -37,12 +38,17 @@ func (c *command) applicationInstanceList(cmd *cobra.Command, _ []string) error
 		return err
 	}
 
+	pageSize, err := getPageSize(cmd)
+	if err != nil {
+		return err
+	}
+
 	client, err := c.GetCmfClient(cmd)
 	if err != nil {
 		return err
 	}
 
-	instances, err := client.ListApplicationInstances(c.createContext(), environment, application)
+	instances, err := client.ListApplicationInstances(c.createContext(), environment, application, pageSize)
 	if err != nil {
 		return err
 	}

internal/flink/command_application_list.go

@@ -18,6 +18,7 @@ func (c *command) newApplicationListCommand() *cobra.Command {
 	}
 
 	cmd.Flags().String("environment", "", "Name of the Flink environment.")
+	addPageSizeFlag(cmd)
 	addCmfFlagSet(cmd)
 	pcmd.AddOutputFlag(cmd)
 
@@ -32,12 +33,17 @@ func (c *command) applicationList(cmd *cobra.Command, _ []string) error {
 		return err
 	}
 
+	pageSize, err := getPageSize(cmd)
+	if err != nil {
+		return err
+	}
+
 	client, err := c.GetCmfClient(cmd)
 	if err != nil {
 		return err
 	}
 
-	applications, err := client.ListApplications(c.createContext(), environment)
+	applications, err := client.ListApplications(c.createContext(), environment, pageSize)
 	if err != nil {
 		return err
 	}

internal/flink/command_catalog_database_list.go

@@ -17,6 +17,7 @@ func (c *command) newCatalogDatabaseListCommand() *cobra.Command {
 
 	cmd.Flags().String("catalog", "", "Name of the catalog.")
 	cobra.CheckErr(cmd.MarkFlagRequired("catalog"))
+	addPageSizeFlag(cmd)
 	addCmfFlagSet(cmd)
 	pcmd.AddOutputFlag(cmd)
 
@@ -29,12 +30,17 @@ func (c *command) catalogDatabaseList(cmd *cobra.Command, _ []string) error {
 		return err
 	}
 
+	pageSize, err := getPageSize(cmd)
+	if err != nil {
+		return err
+	}
+
 	client, err := c.GetCmfClient(cmd)
 	if err != nil {
 		return err
 	}
 
-	sdkDatabases, err := client.ListDatabases(c.createContext(), catalogName)
+	sdkDatabases, err := client.ListDatabases(c.createContext(), catalogName, pageSize)
 	if err != nil {
 		return err
 	}

internal/flink/command_catalog_list.go

@@ -15,19 +15,25 @@ func (c *command) newCatalogListCommand() *cobra.Command {
 		RunE:  c.catalogList,
 	}
 
+	addPageSizeFlag(cmd)
 	addCmfFlagSet(cmd)
 	pcmd.AddOutputFlag(cmd)
 
 	return cmd
 }
 
 func (c *command) catalogList(cmd *cobra.Command, _ []string) error {
+	pageSize, err := getPageSize(cmd)
+	if err != nil {
+		return err
+	}
+
 	client, err := c.GetCmfClient(cmd)
 	if err != nil {
 		return err
 	}
 
-	sdkCatalogs, err := client.ListCatalog(c.createContext())
+	sdkCatalogs, err := client.ListCatalog(c.createContext(), pageSize)
 	if err != nil {
 		return err
 	}

internal/flink/command_compute_pool_list_onprem.go

@@ -17,6 +17,7 @@ func (c *command) newComputePoolListCommandOnPrem() *cobra.Command {
 	}
 
 	cmd.Flags().String("environment", "", "Name of the Flink environment.")
+	addPageSizeFlag(cmd)
 	addCmfFlagSet(cmd)
 	pcmd.AddOutputFlag(cmd)
 	cobra.CheckErr(cmd.MarkFlagRequired("environment"))
@@ -30,12 +31,17 @@ func (c *command) computePoolListOnPrem(cmd *cobra.Command, _ []string) error {
 		return err
 	}
 
+	pageSize, err := getPageSize(cmd)
+	if err != nil {
+		return err
+	}
+
 	client, err := c.GetCmfClient(cmd)
 	if err != nil {
 		return err
 	}
 
-	sdkComputePools, err := client.ListComputePools(c.createContext(), environment)
+	sdkComputePools, err := client.ListComputePools(c.createContext(), environment, pageSize)
 	if err != nil {
 		return err
 	}

internal/flink/command_detached_savepoint_list.go

@@ -23,6 +23,7 @@ func (c *command) newDetachedSavepointListCommand() *cobra.Command {
 	}
 
 	cmd.Flags().String("filter", "", "A filter expression to filter by detached savepoint name prefix.")
+	addPageSizeFlag(cmd)
 
 	pcmd.AddOutputFlag(cmd)
 	addCmfFlagSet(cmd)
@@ -41,7 +42,12 @@ func (c *command) detachedSavepointList(cmd *cobra.Command, args []string) error
 		return err
 	}
 
-	detachedSavepoints, err := client.ListDetachedSavepoint(c.createContext(), filter)
+	pageSize, err := getPageSize(cmd)
+	if err != nil {
+		return err
+	}
+
+	detachedSavepoints, err := client.ListDetachedSavepoint(c.createContext(), filter, pageSize)
 	if err != nil {
 		return err
 	}

internal/flink/command_environment_list.go

@@ -15,6 +15,7 @@ func (c *command) newEnvironmentListCommand() *cobra.Command {
 		RunE:  c.environmentList,
 	}
 
+	addPageSizeFlag(cmd)
 	addCmfFlagSet(cmd)
 
 	pcmd.AddOutputFlag(cmd)
@@ -23,12 +24,17 @@ func (c *command) newEnvironmentListCommand() *cobra.Command {
 }
 
 func (c *command) environmentList(cmd *cobra.Command, _ []string) error {
+	pageSize, err := getPageSize(cmd)
+	if err != nil {
+		return err
+	}
+
 	client, err := c.GetCmfClient(cmd)
 	if err != nil {
 		return err
 	}
 
-	sdkEnvironments, err := client.ListEnvironments(c.createContext())
+	sdkEnvironments, err := client.ListEnvironments(c.createContext(), pageSize)
 	if err != nil {
 		return err
 	}

internal/flink/command_savepoint_list.go

@@ -19,6 +19,7 @@ func (c *command) newSavepointListCommand() *cobra.Command {
 	cmd.Flags().String("environment", "", "Name of the Flink environment.")
 	cmd.Flags().String("application", "", "The name of the Flink application to list the savepoints.")
 	cmd.Flags().String("statement", "", "The name of the Flink statement to list the savepoints.")
+	addPageSizeFlag(cmd)
 	addCmfFlagSet(cmd)
 	pcmd.AddOutputFlag(cmd)
 
@@ -45,12 +46,17 @@ func (c *command) savepointList(cmd *cobra.Command, _ []string) error {
 		return err
 	}
 
+	pageSize, err := getPageSize(cmd)
+	if err != nil {
+		return err
+	}
+
 	client, err := c.GetCmfClient(cmd)
 	if err != nil {
 		return err
 	}
 
-	sdkSavepoints, err := client.ListSavepoint(c.createContext(), environment, statement, application, statement != "")
+	sdkSavepoints, err := client.ListSavepoint(c.createContext(), environment, statement, application, statement != "", pageSize)
 	if err != nil {
 		return err
 	}

internal/flink/command_secret_list.go

@@ -15,19 +15,25 @@ func (c *command) newSecretListCommand() *cobra.Command {
 		RunE:  c.secretList,
 	}
 
+	addPageSizeFlag(cmd)
 	addCmfFlagSet(cmd)
 	pcmd.AddOutputFlag(cmd)
 
 	return cmd
 }
 
 func (c *command) secretList(cmd *cobra.Command, _ []string) error {
+	pageSize, err := getPageSize(cmd)
+	if err != nil {
+		return err
+	}
+
 	client, err := c.GetCmfClient(cmd)
 	if err != nil {
 		return err
 	}
 
-	sdkSecrets, err := client.ListSecrets(c.createContext())
+	sdkSecrets, err := client.ListSecrets(c.createContext(), pageSize)
 	if err != nil {
 		return err
 	}

internal/flink/command_secret_mapping_list.go

@@ -18,6 +18,7 @@ func (c *command) newSecretMappingListCommand() *cobra.Command {
 
 	cmd.Flags().String("environment", "", "Name of the Flink environment.")
 	cobra.CheckErr(cmd.MarkFlagRequired("environment"))
+	addPageSizeFlag(cmd)
 	addCmfFlagSet(cmd)
 	pcmd.AddOutputFlag(cmd)
 
@@ -30,12 +31,17 @@ func (c *command) secretMappingList(cmd *cobra.Command, _ []string) error {
 		return err
 	}
 
+	pageSize, err := getPageSize(cmd)
+	if err != nil {
+		return err
+	}
+
 	client, err := c.GetCmfClient(cmd)
 	if err != nil {
 		return err
 	}
 
-	sdkMappings, err := client.ListSecretMappings(c.createContext(), environment)
+	sdkMappings, err := client.ListSecretMappings(c.createContext(), environment, pageSize)
 	if err != nil {
 		return err
 	}

internal/flink/command_statement_list_onprem.go

@@ -23,6 +23,7 @@ func (c *command) newStatementListCommandOnPrem() *cobra.Command {
 	cmd.Flags().String("environment", "", "Name of the Flink environment.")
 	cmd.Flags().String("compute-pool", "", "Optional flag to filter the Flink statements by compute pool ID.")
 	cmd.Flags().String("status", "", "Optional flag to filter the Flink statements by statement status.")
+	addPageSizeFlag(cmd)
 	addCmfFlagSet(cmd)
 	pcmd.AddOutputFlag(cmd)
 
@@ -57,7 +58,12 @@ func (c *command) statementListOnPrem(cmd *cobra.Command, _ []string) error {
 		return err
 	}
 
-	sdkStatements, err := client.ListStatements(c.createContext(), environment, computePool, status)
+	pageSize, err := getPageSize(cmd)
+	if err != nil {
+		return err
+	}
+
+	sdkStatements, err := client.ListStatements(c.createContext(), environment, computePool, status, pageSize)
 	if err != nil {
 		return err
 	}