fix(ingest/pulsar): handle missing/invalid schema objects

fix(ingest/pulsar): handle missing/invalid schema objects #31310