test: add robust testing to streaming ingestion integration tests

[zhang-arvin]

Description

Adds more robust testing to streaming ingestion integration tests as described in #10971.

Changes

1. New FaultyStreamEventStreamGenerator — A SyntheticStreamGenerator subclass that can inject faulty data at a configurable ratio:

   • INVALID_JSON — Malformed JSON bytes that cannot be parsed (e.g., {"broken": })
   • NULL_FIELDS — Valid JSON objects where all field values are null
   • EMPTY_JSON — Empty JSON objects ({})
   • MULTI_ROW — JSON arrays containing multiple objects
   • EMPTY_STRING — Completely empty strings
   • ALL_VALID — No faults injected (baseline)

2. New KafkaStreamIngestionRobustnessTest — Six integration tests that verify the streaming ingestion pipeline handles faulty data gracefully:

   • test_supervisorHandlesInvalidJsonGracefully — Supervisor stays healthy with 20% invalid JSON
   • test_supervisorHandlesNullFieldsGracefully — Supervisor stays healthy with 20% null-field records
   • test_supervisorHandlesEmptyJsonGracefully — Supervisor stays healthy with 20% empty JSON objects
   • test_supervisorHandlesMultiRowDataGracefully — Supervisor stays healthy with 20% multi-row JSON arrays
   • test_supervisorHandlesEmptyStringGracefully — Supervisor stays healthy with 20% empty strings
   • test_supervisorHandlesAllValidDataCorrectly — Baseline test with all valid data

Each test verifies that:

• The supervisor remains healthy (RUNNING state)
• Valid records are ingested successfully
• The row count matches the expected valid records

Closes #10971.

---

This PR has:

• been self-reviewed.

[FrankChen021]

Severity	Findings
P0	0
P1	0
P2	3
P3	0
Total	3

Reviewed 2 of 2 changed files.

The review found three P2 test-correctness issues involving premature health checks, nondeterministic fault coverage, and an invalid multi-row payload.

---

This is an automated review by Codex GPT-5.6-Luna(max)

[FrankChen021]

[P2] Health is checked before malformed records are consumed

The supervisor is checked before malformed records are consumed; later assertions only wait for valid rows. A supervisor can become unhealthy afterward and the test still passes. Check health after input processing completes.

[FrankChen021]

[P2] Fault injection can emit zero faulty records

Random sampling can emit zero faulty records, while the tests never assert fault coverage. Use deterministic selection or an injected RNG and assert the expected faulty count.

[FrankChen021]

[P2] MULTI_ROW payload is invalid and unsupported

The MULTI_ROW payload has an invalid JSON suffix, so it is rejected as malformed. Kafka's non-line-splittable JsonReader also treats a top-level array as one value, not separate rows. The test only waits for valid rows, so it can pass without testing multi-row ingestion.