From 1ecbfcb8ec40c3d86be70686e2172ea3405734db Mon Sep 17 00:00:00 2001 From: Paras Negi Date: Sun, 2 Aug 2026 14:05:45 +0530 Subject: [PATCH 1/2] [CF-4208] Add --page-size to on-prem Flink list commands The on-prem Flink (CPF/CMF) list commands loop over a hardcoded 100-item page size with no way to control it, so listing a large environment costs many serial round trips (~51 requests for 5000 applications). Add a --page-size flag to every paginated on-prem list command so callers can fetch the full list in fewer, larger requests. It defaults to 100, which preserves today's behavior and output. Collapse the 12 duplicated pagination loops in the CMF client into a single generic listAllPages helper that takes the page size. The requested size is passed through to the CMF page/size API; the loop still terminates on the first empty page. Server-side filtering for `application list` is deliberately left out and will follow in a separate PR stacked on this one. Co-Authored-By: Claude Opus 4.8 --- internal/flink/command.go | 16 ++ .../flink/command_application_event_list.go | 8 +- .../command_application_instance_list.go | 8 +- internal/flink/command_application_list.go | 8 +- .../flink/command_catalog_database_list.go | 8 +- internal/flink/command_catalog_list.go | 8 +- .../flink/command_compute_pool_list_onprem.go | 8 +- .../flink/command_detached_savepoint_list.go | 8 +- internal/flink/command_environment_list.go | 8 +- internal/flink/command_savepoint_list.go | 8 +- internal/flink/command_secret_list.go | 8 +- internal/flink/command_secret_mapping_list.go | 8 +- .../flink/command_statement_list_onprem.go | 8 +- pkg/flink/cmf_rest_client.go | 256 ++++++------------ pkg/flink/cmf_rest_client_test.go | 82 ++++++ pkg/flink/test/mock/cmf_client_mock.go | 8 +- .../application/event-list-app-missing.golden | 1 + .../event-list-missing-flags.golden | 1 + .../application/event/list-help-onprem.golden | 1 + .../instance-list-app-missing.golden | 1 + .../instance-list-env-missing.golden | 1 + .../instance/list-help-onprem.golden | 1 + .../flink/application/list-env-missing.golden | 1 + .../flink/application/list-help-onprem.golden | 1 + .../application/list-page-size-invalid.golden | 1 + .../catalog/database/list-help-onprem.golden | 1 + .../flink/catalog/list-help-onprem.golden | 1 + .../compute-pool/list-help-onprem.golden | 1 + .../list-missing-env-flag-failure.golden | 1 + .../list-help-onprem.golden | 1 + .../flink/environment/list-help-onprem.golden | 1 + .../flink/savepoint/list-fail-both.golden | 1 + .../flink/savepoint/list-help-onprem.golden | 1 + .../secret-mapping/list-help-onprem.golden | 1 + .../flink/secret/list-help-onprem.golden | 1 + .../statement/list-env-missing-failure.golden | 1 + .../flink/statement/list-help-onprem.golden | 1 + test/flink_onprem_test.go | 5 + test/test-server/flink_onprem_handler.go | 47 ++-- 39 files changed, 328 insertions(+), 203 deletions(-) create mode 100644 pkg/flink/cmf_rest_client_test.go create mode 100644 test/fixtures/output/flink/application/list-page-size-invalid.golden diff --git a/internal/flink/command.go b/internal/flink/command.go index 23cf3b6d61..a1336cc1a5 100644 --- a/internal/flink/command.go +++ b/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() diff --git a/internal/flink/command_application_event_list.go b/internal/flink/command_application_event_list.go index 6270f7d7c3..dfc12b426f 100644 --- a/internal/flink/command_application_event_list.go +++ b/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 } diff --git a/internal/flink/command_application_instance_list.go b/internal/flink/command_application_instance_list.go index c3be00d527..25d97deedb 100644 --- a/internal/flink/command_application_instance_list.go +++ b/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 } diff --git a/internal/flink/command_application_list.go b/internal/flink/command_application_list.go index 856ee0fcab..63000b8001 100644 --- a/internal/flink/command_application_list.go +++ b/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 } diff --git a/internal/flink/command_catalog_database_list.go b/internal/flink/command_catalog_database_list.go index 2e0ff82f6e..bfa91b0057 100644 --- a/internal/flink/command_catalog_database_list.go +++ b/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 } diff --git a/internal/flink/command_catalog_list.go b/internal/flink/command_catalog_list.go index b6abf762e3..51c0054b4f 100644 --- a/internal/flink/command_catalog_list.go +++ b/internal/flink/command_catalog_list.go @@ -15,6 +15,7 @@ func (c *command) newCatalogListCommand() *cobra.Command { RunE: c.catalogList, } + addPageSizeFlag(cmd) addCmfFlagSet(cmd) pcmd.AddOutputFlag(cmd) @@ -22,12 +23,17 @@ func (c *command) newCatalogListCommand() *cobra.Command { } 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 } diff --git a/internal/flink/command_compute_pool_list_onprem.go b/internal/flink/command_compute_pool_list_onprem.go index c7a9ff4f71..ba2df5916b 100644 --- a/internal/flink/command_compute_pool_list_onprem.go +++ b/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 } diff --git a/internal/flink/command_detached_savepoint_list.go b/internal/flink/command_detached_savepoint_list.go index 60037382db..22ceabb9d9 100644 --- a/internal/flink/command_detached_savepoint_list.go +++ b/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 } diff --git a/internal/flink/command_environment_list.go b/internal/flink/command_environment_list.go index 26bbd1c362..fdefc70b79 100644 --- a/internal/flink/command_environment_list.go +++ b/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 } diff --git a/internal/flink/command_savepoint_list.go b/internal/flink/command_savepoint_list.go index 287a33d153..fb2ba39998 100644 --- a/internal/flink/command_savepoint_list.go +++ b/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 } diff --git a/internal/flink/command_secret_list.go b/internal/flink/command_secret_list.go index c02c7151b0..88e03a25cd 100644 --- a/internal/flink/command_secret_list.go +++ b/internal/flink/command_secret_list.go @@ -15,6 +15,7 @@ func (c *command) newSecretListCommand() *cobra.Command { RunE: c.secretList, } + addPageSizeFlag(cmd) addCmfFlagSet(cmd) pcmd.AddOutputFlag(cmd) @@ -22,12 +23,17 @@ func (c *command) newSecretListCommand() *cobra.Command { } 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 } diff --git a/internal/flink/command_secret_mapping_list.go b/internal/flink/command_secret_mapping_list.go index 98c9692699..c52763bd44 100644 --- a/internal/flink/command_secret_mapping_list.go +++ b/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 } diff --git a/internal/flink/command_statement_list_onprem.go b/internal/flink/command_statement_list_onprem.go index 1de2836700..8f514f291b 100644 --- a/internal/flink/command_statement_list_onprem.go +++ b/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 } diff --git a/pkg/flink/cmf_rest_client.go b/pkg/flink/cmf_rest_client.go index 8518ad1423..83a76a0672 100644 --- a/pkg/flink/cmf_rest_client.go +++ b/pkg/flink/cmf_rest_client.go @@ -30,7 +30,7 @@ type OnPremCMFRestFlagValues struct { type CmfClientInterface interface { GetStatement(ctx context.Context, environment, name string) (cmfsdk.Statement, error) - ListStatements(ctx context.Context, environment, computePool, status string) ([]cmfsdk.Statement, error) + ListStatements(ctx context.Context, environment, computePool, status string, pageSize int32) ([]cmfsdk.Statement, error) CreateStatement(ctx context.Context, environment string, statement cmfsdk.Statement) (cmfsdk.Statement, error) ListStatementExceptions(ctx context.Context, environment, statementName string) (cmfsdk.StatementExceptionList, error) DeleteStatement(ctx context.Context, environment, statement string) error @@ -189,23 +189,14 @@ func (cmfClient *CmfRestClient) DescribeApplication(ctx context.Context, environ return cmfApplication, nil } -func (cmfClient *CmfRestClient) ListApplications(ctx context.Context, environment string) ([]cmfsdk.FlinkApplication, error) { - applications := make([]cmfsdk.FlinkApplication, 0) - // 100 is an arbitrary page size we've chosen. - var currentPageNumber int32 = 0 - const pageSize = 100 - done := false - - for !done { - applicationsPage, httpResponse, err := cmfClient.FlinkApplicationsApi.GetApplications(ctx, environment).Page(currentPageNumber).Size(pageSize).Execute() +func (cmfClient *CmfRestClient) ListApplications(ctx context.Context, environment string, pageSize int32) ([]cmfsdk.FlinkApplication, error) { + return listAllPages(pageSize, func(page, size int32) ([]cmfsdk.FlinkApplication, error) { + applicationsPage, httpResponse, err := cmfClient.FlinkApplicationsApi.GetApplications(ctx, environment).Page(page).Size(size).Execute() if parsedErr := parseSdkError(httpResponse, err); parsedErr != nil { return nil, fmt.Errorf(`failed to list applications in the environment "%s": %s`, environment, parsedErr) } - applications = append(applications, applicationsPage.GetItems()...) - currentPageNumber, done = extractPageOptions(len(applicationsPage.GetItems()), currentPageNumber) - } - - return applications, nil + return applicationsPage.GetItems(), nil + }) } // UpdateApplication Update an application in the specified environment. @@ -230,22 +221,14 @@ func (cmfClient *CmfRestClient) UpdateApplication(ctx context.Context, environme return outputApplication, nil } -func (cmfClient *CmfRestClient) ListApplicationEvents(ctx context.Context, environment, application string) ([]cmfsdk.FlinkApplicationEvent, error) { - events := make([]cmfsdk.FlinkApplicationEvent, 0) - var currentPageNumber int32 = 0 - const pageSize = 100 - done := false - - for !done { - eventsPage, httpResponse, err := cmfClient.FlinkApplicationsApi.GetApplicationEvents(ctx, environment, application).Page(currentPageNumber).Size(pageSize).Execute() +func (cmfClient *CmfRestClient) ListApplicationEvents(ctx context.Context, environment, application string, pageSize int32) ([]cmfsdk.FlinkApplicationEvent, error) { + return listAllPages(pageSize, func(page, size int32) ([]cmfsdk.FlinkApplicationEvent, error) { + eventsPage, httpResponse, err := cmfClient.FlinkApplicationsApi.GetApplicationEvents(ctx, environment, application).Page(page).Size(size).Execute() if parsedErr := parseSdkError(httpResponse, err); parsedErr != nil { return nil, fmt.Errorf(`failed to list events for application "%s" in the environment "%s": %s`, application, environment, parsedErr) } - events = append(events, eventsPage.GetItems()...) - currentPageNumber, done = extractPageOptions(len(eventsPage.GetItems()), currentPageNumber) - } - - return events, nil + return eventsPage.GetItems(), nil + }) } // CreateEnvironment Create an environment. @@ -280,25 +263,14 @@ func (cmfClient *CmfRestClient) DescribeEnvironment(ctx context.Context, environ return cmfEnvironment, nil } -// ListEnvironments Run through all the pages until we get an empty page, in that case, return. -func (cmfClient *CmfRestClient) ListEnvironments(ctx context.Context) ([]cmfsdk.Environment, error) { - environments := make([]cmfsdk.Environment, 0) - done := false - // 100 is an arbitrary page size we've chosen. - const pageSize = 100 - var currentPageNumber int32 = 0 - - for !done { - environmentsPage, httpResponse, err := cmfClient.EnvironmentsApi.GetEnvironments(ctx).Page(currentPageNumber).Size(pageSize).Execute() +func (cmfClient *CmfRestClient) ListEnvironments(ctx context.Context, pageSize int32) ([]cmfsdk.Environment, error) { + return listAllPages(pageSize, func(page, size int32) ([]cmfsdk.Environment, error) { + environmentsPage, httpResponse, err := cmfClient.EnvironmentsApi.GetEnvironments(ctx).Page(page).Size(size).Execute() if parsedErr := parseSdkError(httpResponse, err); parsedErr != nil { return nil, fmt.Errorf("failed to list environments: %s", parsedErr) } - - environments = append(environments, environmentsPage.GetItems()...) - currentPageNumber, done = extractPageOptions(len(environmentsPage.GetItems()), currentPageNumber) - } - - return environments, nil + return environmentsPage.GetItems(), nil + }) } // UpdateEnvironment updates an existing environment. @@ -371,28 +343,21 @@ func (cmfClient *CmfRestClient) DeleteSavepoint(ctx context.Context, environment } } -func (cmfClient *CmfRestClient) ListSavepoint(ctx context.Context, environment, statement, application string, isStatement bool) ([]cmfsdk.Savepoint, error) { - savepoints := make([]cmfsdk.Savepoint, 0) - done := false - // 100 is an arbitrary page size we've chosen. - const pageSize = 100 - var currentPageNumber int32 = 0 - for !done { +func (cmfClient *CmfRestClient) ListSavepoint(ctx context.Context, environment, statement, application string, isStatement bool, pageSize int32) ([]cmfsdk.Savepoint, error) { + return listAllPages(pageSize, func(page, size int32) ([]cmfsdk.Savepoint, error) { var savepointsPage cmfsdk.SavepointsPage var httpResponse *_nethttp.Response var err error if isStatement { - savepointsPage, httpResponse, err = cmfClient.SavepointsApi.GetSavepointsForFlinkStatement(ctx, environment, statement).Page(currentPageNumber).Size(pageSize).Execute() + savepointsPage, httpResponse, err = cmfClient.SavepointsApi.GetSavepointsForFlinkStatement(ctx, environment, statement).Page(page).Size(size).Execute() } else { - savepointsPage, httpResponse, err = cmfClient.SavepointsApi.GetSavepointsForFlinkApplication(ctx, environment, application).Page(currentPageNumber).Size(pageSize).Execute() + savepointsPage, httpResponse, err = cmfClient.SavepointsApi.GetSavepointsForFlinkApplication(ctx, environment, application).Page(page).Size(size).Execute() } if parsedErr := parseSdkError(httpResponse, err); parsedErr != nil { return nil, fmt.Errorf(`failed to list savepoints in the environment "%s": %s`, environment, parsedErr) } - savepoints = append(savepoints, savepointsPage.GetItems()...) - currentPageNumber, done = extractPageOptions(len(savepointsPage.GetItems()), currentPageNumber) - } - return savepoints, nil + return savepointsPage.GetItems(), nil + }) } func (cmfClient *CmfRestClient) DescribeDetachedSavepoint(ctx context.Context, name string) (cmfsdk.Savepoint, error) { @@ -404,23 +369,14 @@ func (cmfClient *CmfRestClient) DescribeDetachedSavepoint(ctx context.Context, n return detachedSavepoint, nil } -func (cmfClient *CmfRestClient) ListDetachedSavepoint(ctx context.Context, filter string) ([]cmfsdk.Savepoint, error) { - savepoints := make([]cmfsdk.Savepoint, 0) - done := false - // 100 is an arbitrary page size we've chosen. - const pageSize = 100 - var currentPageNumber int32 = 0 - - for !done { - savepointsPage, httpResponse, err := cmfClient.DetachedSavepointsApi.ListDetachedSavepoints(ctx).Page(currentPageNumber).Size(pageSize).Name(filter).Execute() +func (cmfClient *CmfRestClient) ListDetachedSavepoint(ctx context.Context, filter string, pageSize int32) ([]cmfsdk.Savepoint, error) { + return listAllPages(pageSize, func(page, size int32) ([]cmfsdk.Savepoint, error) { + savepointsPage, httpResponse, err := cmfClient.DetachedSavepointsApi.ListDetachedSavepoints(ctx).Page(page).Size(size).Name(filter).Execute() if parsedErr := parseSdkError(httpResponse, err); parsedErr != nil { return nil, fmt.Errorf(`failed to list detached savepoints %s`, parsedErr) } - savepoints = append(savepoints, savepointsPage.GetItems()...) - currentPageNumber, done = extractPageOptions(len(savepointsPage.GetItems()), currentPageNumber) - } - - return savepoints, nil + return savepointsPage.GetItems(), nil + }) } func (cmfClient *CmfRestClient) DeleteDetachedSavepoint(ctx context.Context, name string) error { @@ -453,23 +409,14 @@ func (cmfClient *CmfRestClient) DescribeComputePool(ctx context.Context, environ return cmfComputePool, nil } -func (cmfClient *CmfRestClient) ListComputePools(ctx context.Context, environment string) ([]cmfsdk.ComputePool, error) { - computePools := make([]cmfsdk.ComputePool, 0) - done := false - // 100 is an arbitrary page size we've chosen. - const pageSize = 100 - var currentPageNumber int32 = 0 - - for !done { - computePoolsPage, httpResponse, err := cmfClient.SQLApi.GetComputePools(ctx, environment).Page(currentPageNumber).Size(pageSize).Execute() +func (cmfClient *CmfRestClient) ListComputePools(ctx context.Context, environment string, pageSize int32) ([]cmfsdk.ComputePool, error) { + return listAllPages(pageSize, func(page, size int32) ([]cmfsdk.ComputePool, error) { + computePoolsPage, httpResponse, err := cmfClient.SQLApi.GetComputePools(ctx, environment).Page(page).Size(size).Execute() if parsedErr := parseSdkError(httpResponse, err); parsedErr != nil { return nil, fmt.Errorf(`failed to list compute pools in the environment "%s": %s`, environment, parsedErr) } - computePools = append(computePools, computePoolsPage.GetItems()...) - currentPageNumber, done = extractPageOptions(len(computePoolsPage.GetItems()), currentPageNumber) - } - - return computePools, nil + return computePoolsPage.GetItems(), nil + }) } func (cmfClient *CmfRestClient) CreateStatement(ctx context.Context, environment string, statement cmfsdk.Statement) (cmfsdk.Statement, error) { @@ -502,13 +449,7 @@ func (cmfClient *CmfRestClient) DeleteStatement(ctx context.Context, environment return parseSdkError(httpResp, err) } -func (cmfClient *CmfRestClient) ListStatements(ctx context.Context, environment, computePool, status string) ([]cmfsdk.Statement, error) { - statements := make([]cmfsdk.Statement, 0) - done := false - // 100 is an arbitrary page size we've chosen. - const pageSize = 100 - var currentPageNumber int32 = 0 - +func (cmfClient *CmfRestClient) ListStatements(ctx context.Context, environment, computePool, status string, pageSize int32) ([]cmfsdk.Statement, error) { request := cmfClient.SQLApi.GetStatements(ctx, environment) if computePool != "" { request = request.ComputePool(computePool) @@ -517,16 +458,13 @@ func (cmfClient *CmfRestClient) ListStatements(ctx context.Context, environment, request = request.Phase(status) } - for !done { - statementsPage, httpResponse, err := request.Page(currentPageNumber).Size(pageSize).Execute() + return listAllPages(pageSize, func(page, size int32) ([]cmfsdk.Statement, error) { + statementsPage, httpResponse, err := request.Page(page).Size(size).Execute() if parsedErr := parseSdkError(httpResponse, err); parsedErr != nil { return nil, fmt.Errorf(`failed to list statements in the environment "%s": %s`, environment, parsedErr) } - statements = append(statements, statementsPage.GetItems()...) - currentPageNumber, done = extractPageOptions(len(statementsPage.GetItems()), currentPageNumber) - } - - return statements, nil + return statementsPage.GetItems(), nil + }) } func (cmfClient *CmfRestClient) ListStatementExceptions(ctx context.Context, environment, statementName string) (cmfsdk.StatementExceptionList, error) { @@ -606,23 +544,14 @@ func (cmfClient *CmfRestClient) DescribeCatalog(ctx context.Context, catalogName return outputCatalog, nil } -func (cmfClient *CmfRestClient) ListCatalog(ctx context.Context) ([]cmfsdk.KafkaCatalog, error) { - catalogs := make([]cmfsdk.KafkaCatalog, 0) - done := false - // 100 is an arbitrary page size we've chosen. - const pageSize = 100 - var currentPageNumber int32 = 0 - - for !done { - catalogPage, httpResponse, err := cmfClient.SQLApi.GetKafkaCatalogs(ctx).Page(currentPageNumber).Size(pageSize).Execute() +func (cmfClient *CmfRestClient) ListCatalog(ctx context.Context, pageSize int32) ([]cmfsdk.KafkaCatalog, error) { + return listAllPages(pageSize, func(page, size int32) ([]cmfsdk.KafkaCatalog, error) { + catalogPage, httpResponse, err := cmfClient.SQLApi.GetKafkaCatalogs(ctx).Page(page).Size(size).Execute() if parsedErr := parseSdkError(httpResponse, err); parsedErr != nil { return nil, fmt.Errorf(`failed to list Kafka Catalog: %s`, parsedErr) } - catalogs = append(catalogs, catalogPage.GetItems()...) - currentPageNumber, done = extractPageOptions(len(catalogPage.GetItems()), currentPageNumber) - } - - return catalogs, nil + return catalogPage.GetItems(), nil + }) } func (cmfClient *CmfRestClient) UpdateCatalog(ctx context.Context, catalogName string, kafkaCatalog cmfsdk.KafkaCatalog) error { @@ -646,23 +575,14 @@ func (cmfClient *CmfRestClient) DescribeApplicationInstance(ctx context.Context, return cmfInstance, nil } -func (cmfClient *CmfRestClient) ListApplicationInstances(ctx context.Context, environment, application string) ([]cmfsdk.FlinkApplicationInstance, error) { - instances := make([]cmfsdk.FlinkApplicationInstance, 0) - var currentPageNumber int32 = 0 - // 100 is an arbitrary page size we've chosen. - const pageSize = 100 - done := false - - for !done { - instancesPage, httpResponse, err := cmfClient.FlinkApplicationsApi.GetApplicationInstances(ctx, environment, application).Page(currentPageNumber).Size(pageSize).Execute() +func (cmfClient *CmfRestClient) ListApplicationInstances(ctx context.Context, environment, application string, pageSize int32) ([]cmfsdk.FlinkApplicationInstance, error) { + return listAllPages(pageSize, func(page, size int32) ([]cmfsdk.FlinkApplicationInstance, error) { + instancesPage, httpResponse, err := cmfClient.FlinkApplicationsApi.GetApplicationInstances(ctx, environment, application).Page(page).Size(size).Execute() if parsedErr := parseSdkError(httpResponse, err); parsedErr != nil { return nil, fmt.Errorf(`failed to list instances of application "%s" in the environment "%s": %s`, application, environment, parsedErr) } - instances = append(instances, instancesPage.GetItems()...) - currentPageNumber, done = extractPageOptions(len(instancesPage.GetItems()), currentPageNumber) - } - - return instances, nil + return instancesPage.GetItems(), nil + }) } func (cmfClient *CmfRestClient) CreateSecretMapping(ctx context.Context, envName string, secretMapping cmfsdk.EnvironmentSecretMapping) (cmfsdk.EnvironmentSecretMapping, error) { @@ -685,22 +605,14 @@ func (cmfClient *CmfRestClient) DescribeSecretMapping(ctx context.Context, envNa return outputMapping, nil } -func (cmfClient *CmfRestClient) ListSecretMappings(ctx context.Context, envName string) ([]cmfsdk.EnvironmentSecretMapping, error) { - mappings := make([]cmfsdk.EnvironmentSecretMapping, 0) - done := false - const pageSize = 100 - var currentPageNumber int32 = 0 - - for !done { - mappingsPage, httpResponse, err := cmfClient.EnvironmentsApi.GetEnvironmentSecretMappings(ctx, envName).Page(currentPageNumber).Size(pageSize).Execute() +func (cmfClient *CmfRestClient) ListSecretMappings(ctx context.Context, envName string, pageSize int32) ([]cmfsdk.EnvironmentSecretMapping, error) { + return listAllPages(pageSize, func(page, size int32) ([]cmfsdk.EnvironmentSecretMapping, error) { + mappingsPage, httpResponse, err := cmfClient.EnvironmentsApi.GetEnvironmentSecretMappings(ctx, envName).Page(page).Size(size).Execute() if parsedErr := parseSdkError(httpResponse, err); parsedErr != nil { return nil, fmt.Errorf(`failed to list secret mappings in the environment "%s": %s`, envName, parsedErr) } - mappings = append(mappings, mappingsPage.GetItems()...) - currentPageNumber, done = extractPageOptions(len(mappingsPage.GetItems()), currentPageNumber) - } - - return mappings, nil + return mappingsPage.GetItems(), nil + }) } func (cmfClient *CmfRestClient) UpdateSecretMapping(ctx context.Context, envName, name string, secretMapping cmfsdk.EnvironmentSecretMapping) (cmfsdk.EnvironmentSecretMapping, error) { @@ -733,22 +645,14 @@ func (cmfClient *CmfRestClient) DescribeSecret(ctx context.Context, secretName s return outputSecret, nil } -func (cmfClient *CmfRestClient) ListSecrets(ctx context.Context) ([]cmfsdk.Secret, error) { - secrets := make([]cmfsdk.Secret, 0) - done := false - const pageSize = 100 - var currentPageNumber int32 = 0 - - for !done { - secretsPage, httpResponse, err := cmfClient.SecretsApi.GetSecrets(ctx).Page(currentPageNumber).Size(pageSize).Execute() +func (cmfClient *CmfRestClient) ListSecrets(ctx context.Context, pageSize int32) ([]cmfsdk.Secret, error) { + return listAllPages(pageSize, func(page, size int32) ([]cmfsdk.Secret, error) { + secretsPage, httpResponse, err := cmfClient.SecretsApi.GetSecrets(ctx).Page(page).Size(size).Execute() if parsedErr := parseSdkError(httpResponse, err); parsedErr != nil { return nil, fmt.Errorf(`failed to list secrets: %s`, parsedErr) } - secrets = append(secrets, secretsPage.GetItems()...) - currentPageNumber, done = extractPageOptions(len(secretsPage.GetItems()), currentPageNumber) - } - - return secrets, nil + return secretsPage.GetItems(), nil + }) } func (cmfClient *CmfRestClient) UpdateSecret(ctx context.Context, secretName string, secret cmfsdk.Secret) (cmfsdk.Secret, error) { @@ -797,30 +701,42 @@ func (cmfClient *CmfRestClient) DescribeDatabase(ctx context.Context, catalogNam return outputDatabase, nil } -func (cmfClient *CmfRestClient) ListDatabases(ctx context.Context, catalogName string) ([]cmfsdk.KafkaDatabase, error) { - databases := make([]cmfsdk.KafkaDatabase, 0) - done := false - const pageSize = 100 - var currentPageNumber int32 = 0 - - for !done { - databasePage, httpResponse, err := cmfClient.SQLApi.GetKafkaDatabases(ctx, catalogName).Page(currentPageNumber).Size(pageSize).Execute() +func (cmfClient *CmfRestClient) ListDatabases(ctx context.Context, catalogName string, pageSize int32) ([]cmfsdk.KafkaDatabase, error) { + return listAllPages(pageSize, func(page, size int32) ([]cmfsdk.KafkaDatabase, error) { + databasePage, httpResponse, err := cmfClient.SQLApi.GetKafkaDatabases(ctx, catalogName).Page(page).Size(size).Execute() if parsedErr := parseSdkError(httpResponse, err); parsedErr != nil { return nil, fmt.Errorf(`failed to list databases in catalog "%s": %s`, catalogName, parsedErr) } - databases = append(databases, databasePage.GetItems()...) - currentPageNumber, done = extractPageOptions(len(databasePage.GetItems()), currentPageNumber) + return databasePage.GetItems(), nil + }) +} + +// listAllPages collects items across all pages by repeatedly calling fetchPage until an empty +// page is returned. pageSize sets the number of items requested per page; a pageSize <= 0 falls +// back to the default of 100. fetchPage receives the zero-based page number and the page size. +func listAllPages[T any](pageSize int32, fetchPage func(page, size int32) ([]T, error)) ([]T, error) { + items := make([]T, 0) + // 100 is an arbitrary default page size we've chosen. + const defaultPageSize int32 = 100 + + size := pageSize + if size <= 0 { + size = defaultPageSize } - return databases, nil -} + for page := int32(0); ; page++ { + pageItems, err := fetchPage(page, size) + if err != nil { + return nil, err + } + items = append(items, pageItems...) -// Returns the next page number and whether we need to fetch more pages or not. -func extractPageOptions(receivedItemsLength int, currentPageNumber int32) (int32, bool) { - if receivedItemsLength == 0 { - return currentPageNumber, true + if len(pageItems) == 0 { + break + } } - return currentPageNumber + 1, false + + return items, nil } // Creates a rich error message from the HTTP response and the SDK error if possible. diff --git a/pkg/flink/cmf_rest_client_test.go b/pkg/flink/cmf_rest_client_test.go new file mode 100644 index 0000000000..689d2d5509 --- /dev/null +++ b/pkg/flink/cmf_rest_client_test.go @@ -0,0 +1,82 @@ +package flink + +import ( + "fmt" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestListAllPages(t *testing.T) { + // pagedFetcher emulates a CMF endpoint that pages by zero-based index (offset = page * size). + // It records the sizes it was asked for so tests can assert on the requested page size. + pagedFetcher := func(total int, requestedSizes *[]int32) func(page, size int32) ([]int, error) { + return func(page, size int32) ([]int, error) { + *requestedSizes = append(*requestedSizes, size) + start := int(page * size) + if start >= total { + return []int{}, nil + } + end := start + int(size) + if end > total { + end = total + } + items := make([]int, 0, end-start) + for i := start; i < end; i++ { + items = append(items, i) + } + return items, nil + } + } + + t.Run("page size 0 defaults to 100 and fetches all pages", func(t *testing.T) { + var sizes []int32 + items, err := listAllPages(0, pagedFetcher(250, &sizes)) + require.NoError(t, err) + require.Len(t, items, 250) + // Requests all use the default size of 100: three carry items (100 + 100 + 50) and a + // final empty page terminates the loop. + require.Equal(t, []int32{100, 100, 100, 100}, sizes) + }) + + t.Run("custom page size controls request size and round-trip count", func(t *testing.T) { + var sizes []int32 + items, err := listAllPages(50, pagedFetcher(100, &sizes)) + require.NoError(t, err) + require.Len(t, items, 100) + // size=50 over 100 items → two full pages then a terminating empty page. + require.Equal(t, []int32{50, 50, 50}, sizes) + }) + + t.Run("larger page size means fewer round trips for the same data", func(t *testing.T) { + var sizes []int32 + items, err := listAllPages(1000, pagedFetcher(250, &sizes)) + require.NoError(t, err) + require.Len(t, items, 250) + // One data page of up to 1000 covers all 250, then a terminating empty page. + require.Equal(t, []int32{1000, 1000}, sizes) + }) + + t.Run("page size larger than total returns all items in one data page", func(t *testing.T) { + var sizes []int32 + items, err := listAllPages(100, pagedFetcher(30, &sizes)) + require.NoError(t, err) + require.Len(t, items, 30) + require.Equal(t, []int32{100, 100}, sizes) + }) + + t.Run("empty result set", func(t *testing.T) { + var sizes []int32 + items, err := listAllPages(0, pagedFetcher(0, &sizes)) + require.NoError(t, err) + require.Empty(t, items) + }) + + t.Run("propagates fetch error", func(t *testing.T) { + wantErr := fmt.Errorf("boom") + _, err := listAllPages(0, func(page, size int32) ([]int, error) { + return nil, wantErr + }) + require.ErrorIs(t, err, wantErr) + }) +} diff --git a/pkg/flink/test/mock/cmf_client_mock.go b/pkg/flink/test/mock/cmf_client_mock.go index dfe5fc37b1..c9b7aa3524 100644 --- a/pkg/flink/test/mock/cmf_client_mock.go +++ b/pkg/flink/test/mock/cmf_client_mock.go @@ -145,18 +145,18 @@ func (mr *MockCmfClientInterfaceMockRecorder) ListStatementExceptions(ctx, envir } // ListStatements mocks base method. -func (m *MockCmfClientInterface) ListStatements(ctx context.Context, environment, computePool, status string) ([]v1.Statement, error) { +func (m *MockCmfClientInterface) ListStatements(ctx context.Context, environment, computePool, status string, pageSize int32) ([]v1.Statement, error) { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "ListStatements", ctx, environment, computePool, status) + ret := m.ctrl.Call(m, "ListStatements", ctx, environment, computePool, status, pageSize) ret0, _ := ret[0].([]v1.Statement) ret1, _ := ret[1].(error) return ret0, ret1 } // ListStatements indicates an expected call of ListStatements. -func (mr *MockCmfClientInterfaceMockRecorder) ListStatements(ctx, environment, computePool, status any) *gomock.Call { +func (mr *MockCmfClientInterfaceMockRecorder) ListStatements(ctx, environment, computePool, status, pageSize any) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListStatements", reflect.TypeOf((*MockCmfClientInterface)(nil).ListStatements), ctx, environment, computePool, status) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListStatements", reflect.TypeOf((*MockCmfClientInterface)(nil).ListStatements), ctx, environment, computePool, status, pageSize) } // UpdateStatement mocks base method. diff --git a/test/fixtures/output/flink/application/event-list-app-missing.golden b/test/fixtures/output/flink/application/event-list-app-missing.golden index 7f894127c6..17a0deb826 100644 --- a/test/fixtures/output/flink/application/event-list-app-missing.golden +++ b/test/fixtures/output/flink/application/event-list-app-missing.golden @@ -5,6 +5,7 @@ Usage: Flags: --environment string REQUIRED: Name of the Flink environment. --application string REQUIRED: Name of the Flink application. + --page-size int Number of results to fetch per API request. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/application/event-list-missing-flags.golden b/test/fixtures/output/flink/application/event-list-missing-flags.golden index a282b8202a..f0c3f83227 100644 --- a/test/fixtures/output/flink/application/event-list-missing-flags.golden +++ b/test/fixtures/output/flink/application/event-list-missing-flags.golden @@ -5,6 +5,7 @@ Usage: Flags: --environment string REQUIRED: Name of the Flink environment. --application string REQUIRED: Name of the Flink application. + --page-size int Number of results to fetch per API request. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/application/event/list-help-onprem.golden b/test/fixtures/output/flink/application/event/list-help-onprem.golden index 6c301ae514..cad157ff21 100644 --- a/test/fixtures/output/flink/application/event/list-help-onprem.golden +++ b/test/fixtures/output/flink/application/event/list-help-onprem.golden @@ -6,6 +6,7 @@ Usage: Flags: --environment string REQUIRED: Name of the Flink environment. --application string REQUIRED: Name of the Flink application. + --page-size int Number of results to fetch per API request. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/application/instance-list-app-missing.golden b/test/fixtures/output/flink/application/instance-list-app-missing.golden index 589416e105..c06f757ad8 100644 --- a/test/fixtures/output/flink/application/instance-list-app-missing.golden +++ b/test/fixtures/output/flink/application/instance-list-app-missing.golden @@ -5,6 +5,7 @@ Usage: Flags: --environment string REQUIRED: Name of the Flink environment. --application string REQUIRED: Name of the Flink application. + --page-size int Number of results to fetch per API request. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/application/instance-list-env-missing.golden b/test/fixtures/output/flink/application/instance-list-env-missing.golden index 876706e597..6896520d60 100644 --- a/test/fixtures/output/flink/application/instance-list-env-missing.golden +++ b/test/fixtures/output/flink/application/instance-list-env-missing.golden @@ -5,6 +5,7 @@ Usage: Flags: --environment string REQUIRED: Name of the Flink environment. --application string REQUIRED: Name of the Flink application. + --page-size int Number of results to fetch per API request. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/application/instance/list-help-onprem.golden b/test/fixtures/output/flink/application/instance/list-help-onprem.golden index e3476bb621..934321b3a1 100644 --- a/test/fixtures/output/flink/application/instance/list-help-onprem.golden +++ b/test/fixtures/output/flink/application/instance/list-help-onprem.golden @@ -6,6 +6,7 @@ Usage: Flags: --environment string REQUIRED: Name of the Flink environment. --application string REQUIRED: Name of the Flink application. + --page-size int Number of results to fetch per API request. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/application/list-env-missing.golden b/test/fixtures/output/flink/application/list-env-missing.golden index 5d48184c17..8358c03deb 100644 --- a/test/fixtures/output/flink/application/list-env-missing.golden +++ b/test/fixtures/output/flink/application/list-env-missing.golden @@ -4,6 +4,7 @@ Usage: Flags: --environment string REQUIRED: Name of the Flink environment. + --page-size int Number of results to fetch per API request. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/application/list-help-onprem.golden b/test/fixtures/output/flink/application/list-help-onprem.golden index 8d6cf45615..750da75330 100644 --- a/test/fixtures/output/flink/application/list-help-onprem.golden +++ b/test/fixtures/output/flink/application/list-help-onprem.golden @@ -5,6 +5,7 @@ Usage: Flags: --environment string REQUIRED: Name of the Flink environment. + --page-size int Number of results to fetch per API request. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/application/list-page-size-invalid.golden b/test/fixtures/output/flink/application/list-page-size-invalid.golden new file mode 100644 index 0000000000..e6f6a3b52e --- /dev/null +++ b/test/fixtures/output/flink/application/list-page-size-invalid.golden @@ -0,0 +1 @@ +Error: `--page-size` must be between 0 and 2147483647 diff --git a/test/fixtures/output/flink/catalog/database/list-help-onprem.golden b/test/fixtures/output/flink/catalog/database/list-help-onprem.golden index 4dbf71ed16..74ee07f6e5 100644 --- a/test/fixtures/output/flink/catalog/database/list-help-onprem.golden +++ b/test/fixtures/output/flink/catalog/database/list-help-onprem.golden @@ -5,6 +5,7 @@ Usage: Flags: --catalog string REQUIRED: Name of the catalog. + --page-size int Number of results to fetch per API request. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/catalog/list-help-onprem.golden b/test/fixtures/output/flink/catalog/list-help-onprem.golden index 720363cc19..eb3afc2ea3 100644 --- a/test/fixtures/output/flink/catalog/list-help-onprem.golden +++ b/test/fixtures/output/flink/catalog/list-help-onprem.golden @@ -4,6 +4,7 @@ Usage: confluent flink catalog list [flags] Flags: + --page-size int Number of results to fetch per API request. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/compute-pool/list-help-onprem.golden b/test/fixtures/output/flink/compute-pool/list-help-onprem.golden index 0873000661..af69cade99 100644 --- a/test/fixtures/output/flink/compute-pool/list-help-onprem.golden +++ b/test/fixtures/output/flink/compute-pool/list-help-onprem.golden @@ -5,6 +5,7 @@ Usage: Flags: --environment string REQUIRED: Name of the Flink environment. + --page-size int Number of results to fetch per API request. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/compute-pool/list-missing-env-flag-failure.golden b/test/fixtures/output/flink/compute-pool/list-missing-env-flag-failure.golden index b197499e76..541462f5ee 100644 --- a/test/fixtures/output/flink/compute-pool/list-missing-env-flag-failure.golden +++ b/test/fixtures/output/flink/compute-pool/list-missing-env-flag-failure.golden @@ -4,6 +4,7 @@ Usage: Flags: --environment string REQUIRED: Name of the Flink environment. + --page-size int Number of results to fetch per API request. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/detached-savepoint/list-help-onprem.golden b/test/fixtures/output/flink/detached-savepoint/list-help-onprem.golden index 98b7c68b91..0f138f2904 100644 --- a/test/fixtures/output/flink/detached-savepoint/list-help-onprem.golden +++ b/test/fixtures/output/flink/detached-savepoint/list-help-onprem.golden @@ -10,6 +10,7 @@ List Flink detached savepoints with filter filter1. Flags: --filter string A filter expression to filter by detached savepoint name prefix. + --page-size int Number of results to fetch per API request. Defaults to 100. -o, --output string Specify the output format as "human", "json", or "yaml". (default "human") --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/environment/list-help-onprem.golden b/test/fixtures/output/flink/environment/list-help-onprem.golden index 413a6ffc75..ae43a601c0 100644 --- a/test/fixtures/output/flink/environment/list-help-onprem.golden +++ b/test/fixtures/output/flink/environment/list-help-onprem.golden @@ -4,6 +4,7 @@ Usage: confluent flink environment list [flags] Flags: + --page-size int Number of results to fetch per API request. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/savepoint/list-fail-both.golden b/test/fixtures/output/flink/savepoint/list-fail-both.golden index 20a5dd155e..74324c7ba9 100644 --- a/test/fixtures/output/flink/savepoint/list-fail-both.golden +++ b/test/fixtures/output/flink/savepoint/list-fail-both.golden @@ -6,6 +6,7 @@ Flags: --environment string REQUIRED: Name of the Flink environment. --application string The name of the Flink application to list the savepoints. --statement string The name of the Flink statement to list the savepoints. + --page-size int Number of results to fetch per API request. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/savepoint/list-help-onprem.golden b/test/fixtures/output/flink/savepoint/list-help-onprem.golden index 4c5071685c..b6455ba9da 100644 --- a/test/fixtures/output/flink/savepoint/list-help-onprem.golden +++ b/test/fixtures/output/flink/savepoint/list-help-onprem.golden @@ -7,6 +7,7 @@ Flags: --environment string REQUIRED: Name of the Flink environment. --application string The name of the Flink application to list the savepoints. --statement string The name of the Flink statement to list the savepoints. + --page-size int Number of results to fetch per API request. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/secret-mapping/list-help-onprem.golden b/test/fixtures/output/flink/secret-mapping/list-help-onprem.golden index 4cc47721ae..293755e4e5 100644 --- a/test/fixtures/output/flink/secret-mapping/list-help-onprem.golden +++ b/test/fixtures/output/flink/secret-mapping/list-help-onprem.golden @@ -5,6 +5,7 @@ Usage: Flags: --environment string REQUIRED: Name of the Flink environment. + --page-size int Number of results to fetch per API request. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/secret/list-help-onprem.golden b/test/fixtures/output/flink/secret/list-help-onprem.golden index c9b6c95e8a..b0a426af9c 100644 --- a/test/fixtures/output/flink/secret/list-help-onprem.golden +++ b/test/fixtures/output/flink/secret/list-help-onprem.golden @@ -4,6 +4,7 @@ Usage: confluent flink secret list [flags] Flags: + --page-size int Number of results to fetch per API request. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/statement/list-env-missing-failure.golden b/test/fixtures/output/flink/statement/list-env-missing-failure.golden index 876b760432..2b503f8ad9 100644 --- a/test/fixtures/output/flink/statement/list-env-missing-failure.golden +++ b/test/fixtures/output/flink/statement/list-env-missing-failure.golden @@ -6,6 +6,7 @@ Flags: --environment string REQUIRED: Name of the Flink environment. --compute-pool string Optional flag to filter the Flink statements by compute pool ID. --status string Optional flag to filter the Flink statements by statement status. + --page-size int Number of results to fetch per API request. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/statement/list-help-onprem.golden b/test/fixtures/output/flink/statement/list-help-onprem.golden index 35093a3e38..dc61fb5de9 100644 --- a/test/fixtures/output/flink/statement/list-help-onprem.golden +++ b/test/fixtures/output/flink/statement/list-help-onprem.golden @@ -7,6 +7,7 @@ Flags: --environment string REQUIRED: Name of the Flink environment. --compute-pool string Optional flag to filter the Flink statements by compute pool ID. --status string Optional flag to filter the Flink statements by statement status. + --page-size int Number of results to fetch per API request. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/flink_onprem_test.go b/test/flink_onprem_test.go index c98ca8c4b6..c7643858c1 100644 --- a/test/flink_onprem_test.go +++ b/test/flink_onprem_test.go @@ -29,11 +29,14 @@ func (s *CLITestSuite) TestFlinkApplicationList() { // failure scenarios {args: "flink application list", fixture: "flink/application/list-env-missing.golden", exitCode: 1}, {args: "flink application list --environment non-existent", fixture: "flink/application/list-non-existent-env.golden", exitCode: 1}, + {args: "flink application list --environment default --page-size -1", fixture: "flink/application/list-page-size-invalid.golden", exitCode: 1}, // success scenarios {args: "flink application list --environment test", fixture: "flink/application/list-empty-env.golden"}, {args: "flink application list --environment default --output json", fixture: "flink/application/list-json.golden"}, {args: "flink application list --environment default --output yaml", fixture: "flink/application/list-yaml.golden"}, {args: "flink application list --environment default --output human", fixture: "flink/application/list-human.golden"}, + // pagination: a small page size still returns the full list, fetched across multiple round trips + {args: "flink application list --environment default --page-size 2 --output json", fixture: "flink/application/list-json.golden"}, } runIntegrationTestsWithMultipleAuth(s, tests) @@ -195,6 +198,7 @@ func (s *CLITestSuite) TestFlinkDetachedSavepointList() { {args: "flink detached-savepoint list", fixture: "flink/detached-savepoint/list-successful.golden"}, {args: "flink detached-savepoint list --output json", fixture: "flink/detached-savepoint/list-successful-json.golden"}, {args: "flink detached-savepoint list --output yaml", fixture: "flink/detached-savepoint/list-successful-yaml.golden"}, + {args: "flink detached-savepoint list --page-size 1 --output json", fixture: "flink/detached-savepoint/list-successful-json.golden"}, } runIntegrationTestsWithMultipleAuth(s, tests) @@ -622,6 +626,7 @@ func (s *CLITestSuite) TestFlinkStatementListOnPrem() { {args: "flink statement list --environment default", fixture: "flink/statement/list-success.golden"}, {args: "flink statement list --environment default -o json", fixture: "flink/statement/list-success-json.golden"}, {args: "flink statement list --environment default -o yaml", fixture: "flink/statement/list-success-yaml.golden"}, + {args: "flink statement list --environment default --page-size 2 -o json", fixture: "flink/statement/list-success-json.golden"}, // failure {args: "flink statement list", fixture: "flink/statement/list-env-missing-failure.golden", exitCode: 1}, {args: "flink statement list --environment non-exist", fixture: "flink/statement/list-non-exist-env-failure.golden", exitCode: 1}, diff --git a/test/test-server/flink_onprem_handler.go b/test/test-server/flink_onprem_handler.go index d58b006ee8..655a600d78 100644 --- a/test/test-server/flink_onprem_handler.go +++ b/test/test-server/flink_onprem_handler.go @@ -6,6 +6,7 @@ import ( "io" "net/http" "os" + "strconv" "strings" "testing" "time" @@ -125,6 +126,26 @@ func createApplication(name string) cmfsdk.FlinkApplication { } } +// paginateApplications emulates the CMF applications endpoint's zero-based page/size paging +// (offset = page * size) so that --page-size is exercised end-to-end. A size <= 0 falls back +// to 100. +func paginateApplications(all []cmfsdk.FlinkApplication, pageParam, sizeParam string) []cmfsdk.FlinkApplication { + page, _ := strconv.Atoi(pageParam) + size, err := strconv.Atoi(sizeParam) + if err != nil || size <= 0 { + size = 100 + } + start := page * size + if start >= len(all) { + return []cmfsdk.FlinkApplication{} + } + end := start + size + if end > len(all) { + end = len(all) + } + return all[start:end] +} + // Helper function to create a Flink environment. func createEnvironment(name string, namespace string) cmfsdk.Environment { createdTime := time.Date(2024, time.September, 10, 23, 0, 0, 0, time.UTC) @@ -694,26 +715,18 @@ func handleCmfApplications(t *testing.T) http.HandlerFunc { } // For the 'test' environment, return an empty list. - // For the 'default' environment, return applications but only on page 0. + // For the 'default' environment, return three applications, paged. // For the 'update-failure' environment, return the 'update-failure-application' application. - applicationsPage := map[string]interface{}{ - "items": []cmfsdk.FlinkApplication{}, + var allItems []cmfsdk.FlinkApplication + switch environment { + case "default": + allItems = []cmfsdk.FlinkApplication{createApplication("default-application-1"), createApplication("default-application-2"), createApplication("default-application-s")} + case "update-failure": + allItems = []cmfsdk.FlinkApplication{createApplication("update-failure-application")} } - page := r.URL.Query().Get("page") - - if environment == "default" && page == "0" { - items := []cmfsdk.FlinkApplication{createApplication("default-application-1"), createApplication("default-application-2"), createApplication("default-application-s")} - applicationsPage = map[string]interface{}{ - "items": items, - } - } - - if environment == "update-failure" && page == "0" { - items := []cmfsdk.FlinkApplication{createApplication("update-failure-application")} - applicationsPage = map[string]interface{}{ - "items": items, - } + applicationsPage := map[string]interface{}{ + "items": paginateApplications(allItems, r.URL.Query().Get("page"), r.URL.Query().Get("size")), } err := json.NewEncoder(w).Encode(applicationsPage) From 600dd7dc1146b8024fc778bc2613a909b9ad398a Mon Sep 17 00:00:00 2001 From: Paras Negi Date: Wed, 5 Aug 2026 11:42:22 +0530 Subject: [PATCH 2/2] [CF-4208] Clarify --page-size help, cover the upper bound, drop tautological pagination cases - Reword the --page-size help so it reads as a per-request batch size, not a cap on total results (regenerate the affected help/usage goldens). - Add an integration case for --page-size > MaxInt32 (3000000000), exercising the previously untested upper-bound branch of getPageSize (reuses the invalid golden). - Drop the statement/detached-savepoint --page-size cases: those mock handlers ignore the size param, so the cases only verified flag plumbing (already covered by the help goldens) rather than pagination. The applications case (mock genuinely pages) plus TestListAllPages remain the real coverage. Co-Authored-By: Claude Opus 4.8 --- internal/flink/command.go | 2 +- .../output/flink/application/event-list-app-missing.golden | 2 +- .../output/flink/application/event-list-missing-flags.golden | 2 +- .../output/flink/application/event/list-help-onprem.golden | 2 +- .../output/flink/application/instance-list-app-missing.golden | 2 +- .../output/flink/application/instance-list-env-missing.golden | 2 +- .../output/flink/application/instance/list-help-onprem.golden | 2 +- test/fixtures/output/flink/application/list-env-missing.golden | 2 +- test/fixtures/output/flink/application/list-help-onprem.golden | 2 +- .../output/flink/catalog/database/list-help-onprem.golden | 2 +- test/fixtures/output/flink/catalog/list-help-onprem.golden | 2 +- .../fixtures/output/flink/compute-pool/list-help-onprem.golden | 2 +- .../flink/compute-pool/list-missing-env-flag-failure.golden | 2 +- .../output/flink/detached-savepoint/list-help-onprem.golden | 2 +- test/fixtures/output/flink/environment/list-help-onprem.golden | 2 +- test/fixtures/output/flink/savepoint/list-fail-both.golden | 2 +- test/fixtures/output/flink/savepoint/list-help-onprem.golden | 2 +- .../output/flink/secret-mapping/list-help-onprem.golden | 2 +- test/fixtures/output/flink/secret/list-help-onprem.golden | 2 +- .../output/flink/statement/list-env-missing-failure.golden | 2 +- test/fixtures/output/flink/statement/list-help-onprem.golden | 2 +- test/flink_onprem_test.go | 3 +-- 22 files changed, 22 insertions(+), 23 deletions(-) diff --git a/internal/flink/command.go b/internal/flink/command.go index a1336cc1a5..e25ba3b937 100644 --- a/internal/flink/command.go +++ b/internal/flink/command.go @@ -127,7 +127,7 @@ func addCmfFlagSet(cmd *cobra.Command) { } func addPageSizeFlag(cmd *cobra.Command) { - cmd.Flags().Int("page-size", 0, "Number of results to fetch per API request. Defaults to 100.") + cmd.Flags().Int("page-size", 0, "Number of results to fetch per API request while paginating; does not cap the total results returned. Defaults to 100.") } func getPageSize(cmd *cobra.Command) (int32, error) { diff --git a/test/fixtures/output/flink/application/event-list-app-missing.golden b/test/fixtures/output/flink/application/event-list-app-missing.golden index 17a0deb826..ae4f9ef3b5 100644 --- a/test/fixtures/output/flink/application/event-list-app-missing.golden +++ b/test/fixtures/output/flink/application/event-list-app-missing.golden @@ -5,7 +5,7 @@ Usage: Flags: --environment string REQUIRED: Name of the Flink environment. --application string REQUIRED: Name of the Flink application. - --page-size int Number of results to fetch per API request. Defaults to 100. + --page-size int Number of results to fetch per API request while paginating; does not cap the total results returned. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/application/event-list-missing-flags.golden b/test/fixtures/output/flink/application/event-list-missing-flags.golden index f0c3f83227..a0ccab39bb 100644 --- a/test/fixtures/output/flink/application/event-list-missing-flags.golden +++ b/test/fixtures/output/flink/application/event-list-missing-flags.golden @@ -5,7 +5,7 @@ Usage: Flags: --environment string REQUIRED: Name of the Flink environment. --application string REQUIRED: Name of the Flink application. - --page-size int Number of results to fetch per API request. Defaults to 100. + --page-size int Number of results to fetch per API request while paginating; does not cap the total results returned. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/application/event/list-help-onprem.golden b/test/fixtures/output/flink/application/event/list-help-onprem.golden index cad157ff21..34a48de250 100644 --- a/test/fixtures/output/flink/application/event/list-help-onprem.golden +++ b/test/fixtures/output/flink/application/event/list-help-onprem.golden @@ -6,7 +6,7 @@ Usage: Flags: --environment string REQUIRED: Name of the Flink environment. --application string REQUIRED: Name of the Flink application. - --page-size int Number of results to fetch per API request. Defaults to 100. + --page-size int Number of results to fetch per API request while paginating; does not cap the total results returned. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/application/instance-list-app-missing.golden b/test/fixtures/output/flink/application/instance-list-app-missing.golden index c06f757ad8..7312d76472 100644 --- a/test/fixtures/output/flink/application/instance-list-app-missing.golden +++ b/test/fixtures/output/flink/application/instance-list-app-missing.golden @@ -5,7 +5,7 @@ Usage: Flags: --environment string REQUIRED: Name of the Flink environment. --application string REQUIRED: Name of the Flink application. - --page-size int Number of results to fetch per API request. Defaults to 100. + --page-size int Number of results to fetch per API request while paginating; does not cap the total results returned. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/application/instance-list-env-missing.golden b/test/fixtures/output/flink/application/instance-list-env-missing.golden index 6896520d60..65dc3690ca 100644 --- a/test/fixtures/output/flink/application/instance-list-env-missing.golden +++ b/test/fixtures/output/flink/application/instance-list-env-missing.golden @@ -5,7 +5,7 @@ Usage: Flags: --environment string REQUIRED: Name of the Flink environment. --application string REQUIRED: Name of the Flink application. - --page-size int Number of results to fetch per API request. Defaults to 100. + --page-size int Number of results to fetch per API request while paginating; does not cap the total results returned. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/application/instance/list-help-onprem.golden b/test/fixtures/output/flink/application/instance/list-help-onprem.golden index 934321b3a1..391f79636e 100644 --- a/test/fixtures/output/flink/application/instance/list-help-onprem.golden +++ b/test/fixtures/output/flink/application/instance/list-help-onprem.golden @@ -6,7 +6,7 @@ Usage: Flags: --environment string REQUIRED: Name of the Flink environment. --application string REQUIRED: Name of the Flink application. - --page-size int Number of results to fetch per API request. Defaults to 100. + --page-size int Number of results to fetch per API request while paginating; does not cap the total results returned. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/application/list-env-missing.golden b/test/fixtures/output/flink/application/list-env-missing.golden index 8358c03deb..9624dc8276 100644 --- a/test/fixtures/output/flink/application/list-env-missing.golden +++ b/test/fixtures/output/flink/application/list-env-missing.golden @@ -4,7 +4,7 @@ Usage: Flags: --environment string REQUIRED: Name of the Flink environment. - --page-size int Number of results to fetch per API request. Defaults to 100. + --page-size int Number of results to fetch per API request while paginating; does not cap the total results returned. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/application/list-help-onprem.golden b/test/fixtures/output/flink/application/list-help-onprem.golden index 750da75330..6a475526d9 100644 --- a/test/fixtures/output/flink/application/list-help-onprem.golden +++ b/test/fixtures/output/flink/application/list-help-onprem.golden @@ -5,7 +5,7 @@ Usage: Flags: --environment string REQUIRED: Name of the Flink environment. - --page-size int Number of results to fetch per API request. Defaults to 100. + --page-size int Number of results to fetch per API request while paginating; does not cap the total results returned. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/catalog/database/list-help-onprem.golden b/test/fixtures/output/flink/catalog/database/list-help-onprem.golden index 74ee07f6e5..351ced0d43 100644 --- a/test/fixtures/output/flink/catalog/database/list-help-onprem.golden +++ b/test/fixtures/output/flink/catalog/database/list-help-onprem.golden @@ -5,7 +5,7 @@ Usage: Flags: --catalog string REQUIRED: Name of the catalog. - --page-size int Number of results to fetch per API request. Defaults to 100. + --page-size int Number of results to fetch per API request while paginating; does not cap the total results returned. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/catalog/list-help-onprem.golden b/test/fixtures/output/flink/catalog/list-help-onprem.golden index eb3afc2ea3..c266fdd11b 100644 --- a/test/fixtures/output/flink/catalog/list-help-onprem.golden +++ b/test/fixtures/output/flink/catalog/list-help-onprem.golden @@ -4,7 +4,7 @@ Usage: confluent flink catalog list [flags] Flags: - --page-size int Number of results to fetch per API request. Defaults to 100. + --page-size int Number of results to fetch per API request while paginating; does not cap the total results returned. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/compute-pool/list-help-onprem.golden b/test/fixtures/output/flink/compute-pool/list-help-onprem.golden index af69cade99..32735d247e 100644 --- a/test/fixtures/output/flink/compute-pool/list-help-onprem.golden +++ b/test/fixtures/output/flink/compute-pool/list-help-onprem.golden @@ -5,7 +5,7 @@ Usage: Flags: --environment string REQUIRED: Name of the Flink environment. - --page-size int Number of results to fetch per API request. Defaults to 100. + --page-size int Number of results to fetch per API request while paginating; does not cap the total results returned. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/compute-pool/list-missing-env-flag-failure.golden b/test/fixtures/output/flink/compute-pool/list-missing-env-flag-failure.golden index 541462f5ee..64c6de0737 100644 --- a/test/fixtures/output/flink/compute-pool/list-missing-env-flag-failure.golden +++ b/test/fixtures/output/flink/compute-pool/list-missing-env-flag-failure.golden @@ -4,7 +4,7 @@ Usage: Flags: --environment string REQUIRED: Name of the Flink environment. - --page-size int Number of results to fetch per API request. Defaults to 100. + --page-size int Number of results to fetch per API request while paginating; does not cap the total results returned. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/detached-savepoint/list-help-onprem.golden b/test/fixtures/output/flink/detached-savepoint/list-help-onprem.golden index 0f138f2904..7efef330b9 100644 --- a/test/fixtures/output/flink/detached-savepoint/list-help-onprem.golden +++ b/test/fixtures/output/flink/detached-savepoint/list-help-onprem.golden @@ -10,7 +10,7 @@ List Flink detached savepoints with filter filter1. Flags: --filter string A filter expression to filter by detached savepoint name prefix. - --page-size int Number of results to fetch per API request. Defaults to 100. + --page-size int Number of results to fetch per API request while paginating; does not cap the total results returned. Defaults to 100. -o, --output string Specify the output format as "human", "json", or "yaml". (default "human") --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/environment/list-help-onprem.golden b/test/fixtures/output/flink/environment/list-help-onprem.golden index ae43a601c0..6ef808236c 100644 --- a/test/fixtures/output/flink/environment/list-help-onprem.golden +++ b/test/fixtures/output/flink/environment/list-help-onprem.golden @@ -4,7 +4,7 @@ Usage: confluent flink environment list [flags] Flags: - --page-size int Number of results to fetch per API request. Defaults to 100. + --page-size int Number of results to fetch per API request while paginating; does not cap the total results returned. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/savepoint/list-fail-both.golden b/test/fixtures/output/flink/savepoint/list-fail-both.golden index 74324c7ba9..f8256a9661 100644 --- a/test/fixtures/output/flink/savepoint/list-fail-both.golden +++ b/test/fixtures/output/flink/savepoint/list-fail-both.golden @@ -6,7 +6,7 @@ Flags: --environment string REQUIRED: Name of the Flink environment. --application string The name of the Flink application to list the savepoints. --statement string The name of the Flink statement to list the savepoints. - --page-size int Number of results to fetch per API request. Defaults to 100. + --page-size int Number of results to fetch per API request while paginating; does not cap the total results returned. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/savepoint/list-help-onprem.golden b/test/fixtures/output/flink/savepoint/list-help-onprem.golden index b6455ba9da..d43d2c1a4d 100644 --- a/test/fixtures/output/flink/savepoint/list-help-onprem.golden +++ b/test/fixtures/output/flink/savepoint/list-help-onprem.golden @@ -7,7 +7,7 @@ Flags: --environment string REQUIRED: Name of the Flink environment. --application string The name of the Flink application to list the savepoints. --statement string The name of the Flink statement to list the savepoints. - --page-size int Number of results to fetch per API request. Defaults to 100. + --page-size int Number of results to fetch per API request while paginating; does not cap the total results returned. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/secret-mapping/list-help-onprem.golden b/test/fixtures/output/flink/secret-mapping/list-help-onprem.golden index 293755e4e5..6415acb489 100644 --- a/test/fixtures/output/flink/secret-mapping/list-help-onprem.golden +++ b/test/fixtures/output/flink/secret-mapping/list-help-onprem.golden @@ -5,7 +5,7 @@ Usage: Flags: --environment string REQUIRED: Name of the Flink environment. - --page-size int Number of results to fetch per API request. Defaults to 100. + --page-size int Number of results to fetch per API request while paginating; does not cap the total results returned. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/secret/list-help-onprem.golden b/test/fixtures/output/flink/secret/list-help-onprem.golden index b0a426af9c..995549cce8 100644 --- a/test/fixtures/output/flink/secret/list-help-onprem.golden +++ b/test/fixtures/output/flink/secret/list-help-onprem.golden @@ -4,7 +4,7 @@ Usage: confluent flink secret list [flags] Flags: - --page-size int Number of results to fetch per API request. Defaults to 100. + --page-size int Number of results to fetch per API request while paginating; does not cap the total results returned. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/statement/list-env-missing-failure.golden b/test/fixtures/output/flink/statement/list-env-missing-failure.golden index 2b503f8ad9..53d6fc72f4 100644 --- a/test/fixtures/output/flink/statement/list-env-missing-failure.golden +++ b/test/fixtures/output/flink/statement/list-env-missing-failure.golden @@ -6,7 +6,7 @@ Flags: --environment string REQUIRED: Name of the Flink environment. --compute-pool string Optional flag to filter the Flink statements by compute pool ID. --status string Optional flag to filter the Flink statements by statement status. - --page-size int Number of results to fetch per API request. Defaults to 100. + --page-size int Number of results to fetch per API request while paginating; does not cap the total results returned. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/fixtures/output/flink/statement/list-help-onprem.golden b/test/fixtures/output/flink/statement/list-help-onprem.golden index dc61fb5de9..4b6b160615 100644 --- a/test/fixtures/output/flink/statement/list-help-onprem.golden +++ b/test/fixtures/output/flink/statement/list-help-onprem.golden @@ -7,7 +7,7 @@ Flags: --environment string REQUIRED: Name of the Flink environment. --compute-pool string Optional flag to filter the Flink statements by compute pool ID. --status string Optional flag to filter the Flink statements by statement status. - --page-size int Number of results to fetch per API request. Defaults to 100. + --page-size int Number of results to fetch per API request while paginating; does not cap the total results returned. Defaults to 100. --url string Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag. --client-key-path string Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag. --client-cert-path string Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag. diff --git a/test/flink_onprem_test.go b/test/flink_onprem_test.go index c7643858c1..b5b466f46e 100644 --- a/test/flink_onprem_test.go +++ b/test/flink_onprem_test.go @@ -30,6 +30,7 @@ func (s *CLITestSuite) TestFlinkApplicationList() { {args: "flink application list", fixture: "flink/application/list-env-missing.golden", exitCode: 1}, {args: "flink application list --environment non-existent", fixture: "flink/application/list-non-existent-env.golden", exitCode: 1}, {args: "flink application list --environment default --page-size -1", fixture: "flink/application/list-page-size-invalid.golden", exitCode: 1}, + {args: "flink application list --environment default --page-size 3000000000", fixture: "flink/application/list-page-size-invalid.golden", exitCode: 1}, // success scenarios {args: "flink application list --environment test", fixture: "flink/application/list-empty-env.golden"}, {args: "flink application list --environment default --output json", fixture: "flink/application/list-json.golden"}, @@ -198,7 +199,6 @@ func (s *CLITestSuite) TestFlinkDetachedSavepointList() { {args: "flink detached-savepoint list", fixture: "flink/detached-savepoint/list-successful.golden"}, {args: "flink detached-savepoint list --output json", fixture: "flink/detached-savepoint/list-successful-json.golden"}, {args: "flink detached-savepoint list --output yaml", fixture: "flink/detached-savepoint/list-successful-yaml.golden"}, - {args: "flink detached-savepoint list --page-size 1 --output json", fixture: "flink/detached-savepoint/list-successful-json.golden"}, } runIntegrationTestsWithMultipleAuth(s, tests) @@ -626,7 +626,6 @@ func (s *CLITestSuite) TestFlinkStatementListOnPrem() { {args: "flink statement list --environment default", fixture: "flink/statement/list-success.golden"}, {args: "flink statement list --environment default -o json", fixture: "flink/statement/list-success-json.golden"}, {args: "flink statement list --environment default -o yaml", fixture: "flink/statement/list-success-yaml.golden"}, - {args: "flink statement list --environment default --page-size 2 -o json", fixture: "flink/statement/list-success-json.golden"}, // failure {args: "flink statement list", fixture: "flink/statement/list-env-missing-failure.golden", exitCode: 1}, {args: "flink statement list --environment non-exist", fixture: "flink/statement/list-non-exist-env-failure.golden", exitCode: 1},