fix: Pre-create S3A event log dir before SparkContext init

[abhijeet-dhumal]

What this PR does / why we need it:

When spark.eventLog.enabled: "true" and spark.eventLog.dir points to an S3A path, feast materialize-incremental silently writes nothing to the online store and exits with code 0.
The failure chain:

SparkContext.__init__
  └─ EventLoggingListener.start()
       └─ EventLogFileWriter.requireLogBaseDirAsDirectory()
            └─ S3A 404 (prefix doesn't exist) → raises RuntimeException
                 └─ caught by _materialize_one(except Exception) → ERROR job
                      └─ CLI exits 0 — no data written, no visible error

S3 has no real directories. An empty prefix is indistinguishable from "does not exist", so Spark's pre-flight check always fails on a fresh bucket.

Which issue(s) this PR fixes:

In get_or_create_new_spark_session() (compute_engines/spark/utils.py), before building the SparkSession, call _ensure_s3a_event_log_dir() which:

1. Checks if the S3A prefix already contains objects (no-op if it does)
2. Writes a zero-byte .keep placeholder if empty
3. Uses boto3 — already a Feast dependency via the S3 offline store
4. Is fully non-fatal: swallows errors and lets Spark surface its own message if the write fails

No-ops for non-S3A paths (hdfs://, file://, etc.) and when event logging is disabled.

Checks

• I've made sure the tests are passing.
• My commits are signed off (git commit -s)
• My PR title follows conventional commits format

Testing Strategy

• Unit tests
• Integration tests
• Manual tests
• Testing is not required for this change

Misc

---

[R-behera]

This looks like a useful guard for the S3A event log edge case, and the focused tests help. One follow-up worth considering is whether some Feast users rely on credentials or endpoint details only through Spark/Hadoop config rather than environment variables. If so, a short note or test around that path could prevent surprises when the pre-create step runs before Spark fully applies the config.

[ntkathole]

access_key = spark_config.get(
    "spark.hadoop.fs.s3a.access.key",
    os.environ.get("AWS_ACCESS_KEY_ID", ""),
)
secret_key = spark_config.get(
    "spark.hadoop.fs.s3a.secret.key",
    os.environ.get("AWS_SECRET_ACCESS_KEY", ""),
)
session_token = spark_config.get(
    "spark.hadoop.fs.s3a.session.token",
    os.environ.get("AWS_SESSION_TOKEN", ""),
) or None

[ntkathole]

@abhijeet-dhumal Let's handle both comment from devin and @R-behera suggestion

[abhijeet-dhumal]

@abhijeet-dhumal Let's handle both comment from devin and @R-behera suggestion

@ntkathole Addressed both your comments ✅
credentials (access.key, secret.key, session.token) are now read from spark config first with env var fallback, and the Devin-flagged bucket-root path bug is fixed.

[abhijeet-dhumal]

This looks like a useful guard for the S3A event log edge case, and the focused tests help. One follow-up worth considering is whether some Feast users rely on credentials or endpoint details only through Spark/Hadoop config rather than environment variables. If so, a short note or test around that path could prevent surprises when the pre-create step runs before Spark fully applies the config.

@R-behera Good catch on the Spark/Hadoop config credentials path ✅
_ensure_s3a_event_log_dir now reads spark.hadoop.fs.s3a.access.key, secret.key, and session.token from the spark config before falling back to environment variables. Added tests verifying both the spark-config-takes-precedence and env-var-fallback paths.

[ntkathole]

Wondering if this can be AWS_ENDPOINT_URL instead or atleast we need to document this new env var in our docs ?

[abhijeet-dhumal]

Good call — switched to AWS_ENDPOINT_URL . No custom env vars to document now. Spark config (spark.hadoop.fs.s3a.endpoint) still takes precedence when set.

[ntkathole]

@abhijeet-dhumal let's fix the linting

[ntkathole]

Also, consider supporting minio or other path style

addressing_style = (
    "path"
    if spark_config.get("spark.hadoop.fs.s3a.path.style.access", "false").lower() == "true"
    else "auto"
)

config=BotoConfig(
    signature_version="s3v4",
    s3={"addressing_style": addressing_style},
)

[abhijeet-dhumal]

Added .. - _ensure_s3a_event_log_dir now reads spark.hadoop.fs.s3a.path.style.access and passes addressing_style: "path" to BotoConfig when it's "true", otherwise defaults to "auto". Tests cover both paths