diff --git a/test/e2e/sparkapplication_test.go b/test/e2e/sparkapplication_test.go index ae5531f89..8d943c5e5 100644 --- a/test/e2e/sparkapplication_test.go +++ b/test/e2e/sparkapplication_test.go @@ -265,9 +265,9 @@ var _ = Describe("Example SparkApplication", func() { }) It("Fails submission and retries until retries are exhausted", func() { - By("Waiting for SparkApplication to complete") + By("Waiting for SparkApplication to terminate") key := types.NamespacedName{Namespace: app.Namespace, Name: app.Name} - apps, polling_err := collectStatusesUntilSparkApplicationTerminates(ctx, key) + apps, polling_err := collectSparkApplicationsUntilTermination(ctx, key) Expect(polling_err).To(HaveOccurred()) By("Should eventually fail") @@ -320,9 +320,9 @@ var _ = Describe("Example SparkApplication", func() { }) It("Application fails and retries until retries are exhausted", func() { - By("Waiting for SparkApplication to complete") + By("Waiting for SparkApplication to terminate") key := types.NamespacedName{Namespace: app.Namespace, Name: app.Name} - apps, polling_err := collectStatusesUntilSparkApplicationTerminates(ctx, key) + apps, polling_err := collectSparkApplicationsUntilTermination(ctx, key) Expect(polling_err).To(HaveOccurred()) By("Should eventually fail") diff --git a/test/e2e/suit_test.go b/test/e2e/suit_test.go index 2a4c46bdd..e9c534d94 100644 --- a/test/e2e/suit_test.go +++ b/test/e2e/suit_test.go @@ -262,9 +262,8 @@ func waitForSparkApplicationCompleted(ctx context.Context, key types.NamespacedN return false, err } switch app.Status.AppState.State { - case v1beta2.ApplicationStateFailed: - // TODO: Try combining this case with the one below. - return true, errors.New(app.Status.AppState.ErrorMessage) + case v1beta2.ApplicationStateFailedSubmission, v1beta2.ApplicationStateFailed: + return false, errors.New(app.Status.AppState.ErrorMessage) case v1beta2.ApplicationStateCompleted: return true, nil } @@ -273,13 +272,12 @@ func waitForSparkApplicationCompleted(ctx context.Context, key types.NamespacedN return err } - -func collectStatusesUntilSparkApplicationTerminates(ctx context.Context, key types.NamespacedName) ([]v1beta2.SparkApplication, error) { +func collectSparkApplicationsUntilTermination(ctx context.Context, key types.NamespacedName) ([]v1beta2.SparkApplication, error) { cancelCtx, cancelFunc := context.WithTimeout(ctx, WaitTimeout) defer cancelFunc() apps := []v1beta2.SparkApplication{} - + err := wait.PollUntilContextCancel(cancelCtx, PollInterval, true, func(ctx context.Context) (bool, error) { app := v1beta2.SparkApplication{} if err := k8sClient.Get(ctx, key, &app); err != nil {