[CF-4208] Add --name/--status filtering to on-prem application list

internal/flink/command.go

@@ -125,6 +125,14 @@ 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().Int32("page-size", 100, "Number of results to fetch per API request while paginating; does not cap the total results returned.")
+}
+
+func getPageSize(cmd *cobra.Command) (int32, error) {
+	return cmd.Flags().GetInt32("page-size")
+}
+
 func (c *command) createContext() context.Context {
 	if !c.Config.IsOnPremLogin() {
 		return context.Background()

internal/flink/command_application_event_list.go

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

internal/flink/command_application_instance_list.go

@@ -17,6 +17,7 @@
 
 	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)
 
@@ -26,7 +27,7 @@
 	return cmd
 }
 
 func (c *command) applicationInstanceList(cmd *cobra.Command, _ []string) error {
 	environment, err := cmd.Flags().GetString("environment")
 	if err != nil {
 		return err
@@ -37,12 +38,17 @@
 		return err
 	}
 
+	pageSize, err := getPageSize(cmd)
+	if err != nil {
+		return err
+	}
+
 	client, err := c.GetCmfClient(cmd)
 	if err != nil {
 		return err
 	}
 
-	instances, err := client.ListApplicationInstances(c.createContext(), environment, application)
+	instances, err := client.ListApplicationInstances(c.createContext(), environment, application, pageSize)
 	if err != nil {
 		return err
 	}

internal/flink/command_application_list.go

@@ -2,13 +2,22 @@
 
 import (
 	"fmt"
+	"slices"
+	"strings"
 
 	"github.com/spf13/cobra"
 
 	pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
 	"github.com/confluentinc/cli/v4/pkg/output"
+	"github.com/confluentinc/cli/v4/pkg/utils"
 )
 
+// allowedApplicationStatuses lists the Flink job states recognized by the CMF applications
+// "state=" filter, per the cmf-sdk-go GetApplications filter documentation. Unknown values are
+// still forwarded (the server returns no matches rather than erroring); this list only drives
+// the advisory --status warning.
+var allowedApplicationStatuses = []string{"RUNNING", "FINISHED", "FAILED", "CANCELED", "RECONCILING", "COMPLETED", "UNKNOWN"}
+
 func (c *command) newApplicationListCommand() *cobra.Command {
 	cmd := &cobra.Command{
 		Use:   "list",
@@ -18,6 +27,9 @@
 	}
 
 	cmd.Flags().String("environment", "", "Name of the Flink environment.")
+	cmd.Flags().String("name", "", `Filter the Flink applications by name. Supports wildcards, for example "my-app*".`)
+	cmd.Flags().String("status", "", "Filter the Flink applications by status.")
+	addPageSizeFlag(cmd)
 	addCmfFlagSet(cmd)
 	pcmd.AddOutputFlag(cmd)
 
@@ -26,18 +38,39 @@
 	return cmd
 }
 
 func (c *command) applicationList(cmd *cobra.Command, _ []string) error {
 	environment, err := cmd.Flags().GetString("environment")
 	if err != nil {
 		return err
 	}
 
+	name, err := cmd.Flags().GetString("name")
+	if err != nil {
+		return err
+	}
+
+	status, err := cmd.Flags().GetString("status")
+	if err != nil {
+		return err
+	}
+	if status != "" {
+		status = strings.ToUpper(status)
+		if !slices.Contains(allowedApplicationStatuses, status) {
+			output.ErrPrintf(c.Config.EnableColor, "[WARN] Invalid status %q. Valid statuses are %s.\n", status, utils.ArrayToCommaDelimitedString(allowedApplicationStatuses, "and"))
+		}
+	}
+
+	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, buildApplicationFilter(name, status), pageSize)
 	if err != nil {
 		return err
 	}
@@ -76,3 +109,19 @@
 
 	return output.SerializedOutput(cmd, localApps)
 }
+
+// buildApplicationFilter composes the CMF applications "filter" query from the user-facing
+// --name and --status flags. The grammar (comma-separated "key=value" expressions, "name="
+// with an optional "*" suffix wildcard, and "state=" for status) follows the CMF applications
+// list API. Values are not escaped: Kubernetes application names and Flink states cannot
+// contain "," or "=", so no ambiguity arises. The caller normalizes and warns about status.
+func buildApplicationFilter(name, status string) string {
+	filters := make([]string, 0, 2)
+	if name != "" {
+		filters = append(filters, fmt.Sprintf("name=%s", name))
+	}
+	if status != "" {
+		filters = append(filters, fmt.Sprintf("state=%s", status))
+	}
+	return strings.Join(filters, ",")
+}

internal/flink/command_application_list_test.go

@@ -0,0 +1,28 @@
+package flink
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestBuildApplicationFilter(t *testing.T) {
+	tests := []struct {
+		name   string
+		appn   string
+		status string
+		want   string
+	}{
+		{name: "empty", appn: "", status: "", want: ""},
+		{name: "name only", appn: "my-app", status: "", want: "name=my-app"},
+		{name: "name wildcard", appn: "my-app*", status: "", want: "name=my-app*"},
+		{name: "status only", appn: "", status: "RUNNING", want: "state=RUNNING"},
+		{name: "name and status", appn: "a*", status: "RUNNING", want: "name=a*,state=RUNNING"},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			require.Equal(t, test.want, buildApplicationFilter(test.appn, test.status))
+		})
+	}
+}

internal/flink/command_catalog_database_list.go

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

internal/flink/command_catalog_list.go

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

internal/flink/command_compute_pool_list_onprem.go

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

internal/flink/command_detached_savepoint_list.go

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

internal/flink/command_environment_list.go

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

internal/flink/command_savepoint_list.go

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

internal/flink/command_secret_list.go

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

internal/flink/command_secret_mapping_list.go

@@ -18,24 +18,30 @@
 
 	cmd.Flags().String("environment", "", "Name of the Flink environment.")
 	cobra.CheckErr(cmd.MarkFlagRequired("environment"))
+	addPageSizeFlag(cmd)
 	addCmfFlagSet(cmd)
 	pcmd.AddOutputFlag(cmd)
 
 	return cmd
 }
 
 func (c *command) secretMappingList(cmd *cobra.Command, _ []string) error {
 	environment, err := cmd.Flags().GetString("environment")
 	if err != nil {
 		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
 	}