[Bug] STRING Schema topic sink failed

[ethqunzhong]

I tried to run iceberg sink demo follow with pulsar-io-lakehouse sink docs. it fail commit record because getSchema result unexcepted.

Describe the bug
my test flow shows below:

1. create topic & produce data
   first, I produce lots of data to test topic persistent://public/default/iceberg_test by Flink-connector.
   

message format like:
22061772,1670896138459,mmdc-bigdata-test,11.156.128.57,jobmanager,11.156.128.75,2022-12-13 09:48:58,29
and set topic-schema with bin/pulsar-admin schemas upload command.
therefore,test-topic schema show below:

2. Run the lakehouse sink connector

logs shows sink iceberg failed with schema exception.

there are two question：

1. why getSchemaType result different from these two ways:
   
2. record.getSchema().getSchemaInfo().getSchemaDefinition()=null
   so records will skiped in sinkWriter.run
   
   and i found that in getSchemaDefinition
   
   if SchemaType=STRING/BYTES, it's SchemaDefinition will always be null cause sink failed.

Environment

• Pulsar version: 2.9.3
• Deployment: On-premises cluster
• pulsar-io-lakehouse-connector version: 2.9.3.16

4 broker & 1 function-worker (run as a separate process in separate machines.)

[ethqunzhong]

@hangc0276 @zymap @danpi PTAL.

[danpi]

@hangc0276 @zymap @danpi PTAL.

OK, I will take a look.