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)