Core: Restrict GenericAvroReader class resolution to an explicit allowlist

[damansingh1313]

GenericAvroReader.recordReader used an Avro schema's record name directly as a Java class name, loading and constructing it via reflection with no restriction. Two production paths feed this untrusted bytes/schemas: ManifestFiles.decode and Kafka Connect's AvroUtil.decode. If a schema named a class that happened to already be on the classpath and implement IndexedRecord, that class would be constructed regardless of whether it was ever intended to be reconstructed this way.

This adds a hardcoded allowlist (ALLOWED_RECORD_CLASSES) inside GenericAvroReader and only attempts class resolution for names on that list; anything else now falls back to the existing generic record reader. Kafka Connect's classes are listed by name rather than Class literal since iceberg-core cannot depend on iceberg-kafka-connect-events; a comment cross-references that module's FIELD_ID_TO_CLASS to keep the two from drifting apart.

This keeps the change self-contained to GenericAvroReader, with no changes to AvroEncoderUtil, ManifestFiles, or AvroUtil's method signatures or call sites.

Closes #17802

[uros-b]

A non-null record-class name that is absent from the allowlist falls back to GenericData.Record, mirroring the pre-existing ClassNotFoundException catch. A future caller that routes a new class through AvroEncoderUtil.decode but forgets to add it to ALLOWED_RECORD_CLASSES would get a confusing downstream ClassCastException at the cast site rather than a clear diagnostic.

Perhaps we can do something like throw IllegalArgumentException("Record class not in ALLOWED_RECORD_CLASSES: " + className) when className is non-null but not allowed, instead of falling through.

[damansingh1313]

Dug into this further and wanted to share what I found.

Confirmed the gap is real: I removed GenericManifestFile from the allowlist and ran TestFlinkManifest (its actual production caller via Flink's checkpoint serializer) — it failed with a ClassCastException in DeltaManifestsSerializer, a module away from the real cause, exactly as you'd expect from a silent fallback.

I initially leaned toward throwing an IllegalArgumentException instead of falling back silently, but realized it has a real cost: it makes "class exists on the classpath but isn't allowlisted" an externally observable signal — an attacker could probe arbitrary class names via decode() and use throw-vs-fallback as an oracle for what's present on the classpath. That's counter to the spirit of this issue.

Proposed fix instead: log a WARN naming the class (still falling back to a generic record, same as today) rather than throwing. Gives a clear trail to the real cause without changing anything decode() returns externally, so no new oracle.

Let me know if that trade-off makes sense to you, or if you'd still prefer the throw.

[uros-b]

The Kafka-Connect FQCNs are string literals kept in sync with kafka-connect-events AvroUtil.FIELD_ID_TO_CLASS by comment only (core cannot depend on that module). However, a newly added or renamed Connect event payload, or a core generic type not mirrored here, could potentially degrade to a generic record at decode time (runtime, not compile-time). Something like a cross-module round-trip test in iceberg-kafka-connect could probably enforce the invariant better...

[damansingh1313]

For the current 8 Kafka Connect classes, this is already covered — TestEventSerialization round-trips each one with a strict recursive comparison, so a missing/renamed entry fails CI today (that module's actually in our regular build scope, unlike flink, where we found the real gap above).

The open case is a future class added to kafka-connect-events with no corresponding test — nothing can fully guarantee that at compile time given the module boundary. The WARN log proposed above would catch that too, though: it fires for any allowlist gap, Kafka Connect or otherwise, and names the exact missing class. I'd lean on that rather than a dedicated cross-module test, but happy to add one if you'd still want the extra guarantee.