Fix EMR Serverless connector cold start timeout and script mode outputs

[rohitrsh]

Why

Two bugs in the AWS EMR Serverless connector (added in #3427) surface the first time a task runs against a stopped or newly created application. Neither is caught by the existing unit tests, which only exercise the already-STARTED and Pythonic-mode paths.

Tracking issue: flyteorg/flyte#7286

1. Cold start exceeds the CreateTask deadline

create() called ensure_application_started(), which polls until the application reaches STARTED. A cold EMR Serverless application takes tens of seconds to minutes to start, which is longer than the FlytePropeller CreateTask gRPC deadline. The task failed with DeadlineExceeded before a job run was ever submitted, so the first run against a cold application always failed. A retry then succeeds, because by then the application has started — which makes this look intermittent rather than deterministic.

2. Script mode tasks fail after succeeding

Script and Hive mode tasks declare no Flyte outputs — the entrypoint that writes outputs.pb only runs in Pythonic mode. get() returned a Resource without outputs, so on success FlytePropeller looked for outputs.pb, did not find it, and failed the node:

failed at Node[n0]. OutputsNotFoundError: Outputs not found at [s3://.../n0/data/0/outputs.pb]

The EMR Serverless job itself succeeded, so the execution reported success and failure at the same time.

What changed

Cold start. create() now requests startup without waiting, via a new EMRServerlessHandler.start_application_if_needed() that sends StartApplication and reports whether the application is already STARTED. If it is not, create() returns metadata carrying the prepared StartJobRun request, and get() submits it once the application is ready, reporting RUNNING in the meantime. This moves the blocking wait out of the CreateTask RPC and into the polling loop.

A clientToken is generated in create() and attached to the deferred request, so if get() retries after a partially failed submission, EMR Serverless deduplicates instead of starting a second job run.

delete() handles the deferred state as well, so aborting during startup does not leave an orphaned job run.

Script mode outputs. EMRServerlessJobMetadata now records is_script_mode, and get() returns an empty LiteralMap for succeeded script mode jobs so the node closes out successfully.

CI. Adds flytekit-aws-emr-serverless to the plugin matrix in pythonbuild.yml. The plugin was added in #3427 but never wired into the matrix, so its tests have never run in CI.

Tests

146 passed for the plugin, up from 138. New coverage:

• create() defers submission and issues no StartJobRun when the application is not STARTED
• get() submits the deferred job once STARTED, and reports RUNNING while it is still starting
• get() returns an empty LiteralMap for succeeded script mode jobs, and None for Pythonic mode
• delete() on a deferred job, and on a job that was never submitted
• start_application_if_needed() across already-started, needs-start, transitional and terminal states
• clientToken passthrough in start_job_run

Verified against the versions CI pins (pytest==8.2.1, pytest-asyncio==0.23.7) rather than only local versions, and pre-commit passes with the pinned ruff v0.8.3.

Note for reviewers

delete() currently submits a deferred job before cancelling it when the application has since become STARTED. Returning early without submitting would be cheaper; I kept submit-then-cancel because that is the path validated end to end, but happy to simplify if you would prefer.

ensure_application_started() is left in place but is no longer called by the connector, since create() now uses the non-blocking start_application_if_needed(). I did not remove it because it is a public method on the handler and its own tests still cover it, but I am happy to delete it if you would rather not carry the dead path.

The metadata change is backwards compatible: ResourceMeta is JSON-encoded, and both new fields have defaults, so job metadata written by the currently released version still decodes after a connector upgrade. Verified by round-tripping a legacy payload with neither field present.

The API surface is unchanged — the connector calls exactly the same five EMR Serverless operations as before, in a different order, so no IAM policy change is required to adopt this.