title: How to register custormize serializer for flink kafka format type
body: <p>I create table as below, and the data is from kafka.
  I want to deserialize the json message to Pojo object.
  But the message format is not avro or simple json.
  So I need to know how to register custormized serializer and use it for the 'format.type' property.
  By the way, my flink version is 1.10.0.</p>
  <pre><code>CREATE TABLE MyUserTable(
   uuid VARCHAR,
   orgId VARCHAR
  ) with (
  'connector.type' = 'kafka', 
  'connector.version' = '0.11',
  'connector.topic' = 'topic_name',
  'connector.properties.zookeeper.connect' = 'localhost:2181',
  'connector.properties.bootstrap.servers' = 'localhost:9092',
   'connector.properties.group.id' = 'testGroup',
   'format.type' = 'cutormizeSerializer'
  )
  </code></pre>
  <p>The kafka message body sample, each columnName is the key for Pojo object, and rawData is value:</p>
  <pre><code>{
      &quot;beforeData&quot;: [],
      &quot;byteSize&quot;: 272,
      &quot;columnNumber&quot;: 32,
      &quot;data&quot;: [{
          &quot;byteSize&quot;: 8,
          &quot;columnName&quot;: &quot;APPLY_PERSON_ID&quot;,
          &quot;rawData&quot;: 10017,
          &quot;type&quot;: &quot;LONG&quot;
      }, {
          &quot;byteSize&quot;: 12,
          &quot;columnName&quot;: &quot;UPDATE_SALARY&quot;,
          &quot;rawData&quot;: &quot;11000.000000&quot;,
          &quot;type&quot;: &quot;DOUBLE&quot;
      }, {
          &quot;byteSize&quot;: 11,
          &quot;columnName&quot;: &quot;UP_AMOUNT&quot;,
          &quot;rawData&quot;: &quot;1000.000000&quot;,
          &quot;type&quot;: &quot;DOUBLE&quot;
      }, {
          &quot;byteSize&quot;: 3,
          &quot;columnName&quot;: &quot;CURRENCY&quot;,
          &quot;rawData&quot;: &quot;CNY&quot;,
          &quot;type&quot;: &quot;STRING&quot;
      }, {
          &quot;byteSize&quot;: 32,
          &quot;columnName&quot;: &quot;EXCHANGE_RATE&quot;,
          &quot;rawData&quot;: &quot;1.000000000000000000000000000000&quot;,
          &quot;type&quot;: &quot;DOUBLE&quot;
      },  {
          &quot;byteSize&quot;: 11,
          &quot;columnName&quot;: &quot;DEDUCTED_ACCOUNT&quot;,
          &quot;rawData&quot;: &quot;1000.000000&quot;,
          &quot;type&quot;: &quot;DOUBLE&quot;
      }, {
          &quot;byteSize&quot;: 1,
          &quot;columnName&quot;: &quot;ENTER_AT_PROCESS&quot;,
          &quot;rawData&quot;: &quot;Y&quot;,
          &quot;type&quot;: &quot;STRING&quot;
      }],
      &quot;dataCount&quot;: 0,
      &quot;dataMetaData&quot;: {
          &quot;connector&quot;: &quot;mysql&quot;,
          &quot;pos&quot;: 1000368076,
          &quot;row&quot;: 0,
          &quot;ts_ms&quot;: 1625565737000,
          &quot;snapshot&quot;: &quot;false&quot;,
          &quot;db&quot;: &quot;testdb&quot;,
          &quot;table&quot;: &quot;flow_person_t&quot;
      },
      &quot;key&quot;: &quot;APPLY_PERSON_ID&quot;,
      &quot;memorySize&quot;: 1120,
      &quot;operation&quot;: &quot;insert&quot;,
      &quot;rowIndex&quot;: -1,
      &quot;timestamp&quot;: &quot;1970-01-01 00:00:00&quot;
  }
  </code></pre>
  <p>The Pojo object as below:</p>
  <pre><code>import lombok.Data;
  
  @Data
  public class HrSalaryPersonVO {
      private String uuid;
      private String orgId;
      private String unitId;
      private String effectiveDate;
  
      private int adjustPersonCount;
  
      private Double adjustAmount;
  
      private Double beforeSalaryAmount;
      private Double adjustRate;
  
      private String data0prateType;
  
      private String status;
  }
  </code></pre>