SchemaManager creates invalid table schema if incoming data does not contain all the fields in the existing table.

It incoming data schema is missing a field that was in previous records the schema check at [1] will fail and the system will attempt to create a new schema. This schema will not have fields that are in the existing table but are not in the record. The new schema is generated at [2] and it verifies that all the fields in the selected record are present, it does not validate that all the fields in the existing table are present.

In addition, if there are multiple records with missing or additional fields the schema may not be sufficient to cover all records in the list.

I have a test case [3] on a branch that tests the short schema issue, but does not check for multiple missing fields in the list of records.

[1] https://github.com/Aiven-Open/bigquery-connector-for-apache-kafka/blob/481f9af6749bbae13e86a732b363ab26e5a53b37/kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/SchemaManager.java#L279

[2] https://github.com/Aiven-Open/bigquery-connector-for-apache-kafka/blob/481f9af6749bbae13e86a732b363ab26e5a53b37/kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/SchemaManager.java#L312

[3] https://github.com/Aiven-Open/bigquery-connector-for-apache-kafka/blob/adf10de8e258afefdbd40cd86b503a4dcae16325/kcbq-connector/src/test/java/com/wepay/kafka/connect/bigquery/SchemaManagerTest.java#L786-L807