...

It’s common to have thrift schemas highly nested and keep growing as more product features are added.In order to properly decode/encode thrift payload, we propose a recursive converter from nested thrift struct to Flink Row field. Based on each sub struct class schema type mapping along with field index mapping, this recursive converter could handle very sophisticated nested fields (more than 7 levels deep, 24k characters of schema string) thrift schemacaseTType.STRUCT: returngetRow((TBase) val);publicRowgetRow(TBase tBase) {   List<Map.Entry<?extendsTFieldIdEnum, FieldMetaData>> entries= ….

   // allocate row by thrift field size

...

nested field shall we convert to a row of fields sorted by thrift id.

       if (tBase.isSet(entry.getKey())) {

...

           result.setField(i, getPrimitiveValue(entry.getValue().valueMetaData, val,

...

       } else {

           result.setField(i, getDefaultValue(entry.getValue().valueMetaData,

...

       }

       i++;

   }

 return result;

}

Hive MetaStore dependencies conflict

Hive meta store libraries introduced a specific thrift library version that is not compatible with our internal flink version. This led to our Flink SQL jobs version conflict.  We propose shading  libthrift and fb303 in the flink hive connector, and move those two packages to the project root level maven config; users could place their own customized libthrift jar as well as thrift schema jar into the /lib folder during releaseWe suggest user upgrade hive metastore version to match thrift version given keeping two thrift version leads to hive connector dependencies complicated to manage.

Thrift format table DDL script

...

flink-parent pom.xml

<thrift.version>0.5.0-p6<17</thrift.version>StreamExecutionEnvironmentenv= …

// user can opt-in to skip corrupted state 

env.addDefaultKryoSerializer(Event.class, PinterestTBaseSerializer.class);

...

   topicName,

...

   Configuration.configurationToProperties(kafkaConsumerConfig));

// user can opt-in to deserialize list of fields 

...

   topicName,

...

KafkaSource<Event> source =
        KafkaSource.<Event>builder()
                ...
                .setDeserializer(
                        new ThriftDeserializationSchema<>(Event.class)

...

)
                ...
                .build();

...

   Configuration.configurationToProperties(kafkaConsumerConfig));

...

       Configuration.configurationToProperties(kafkaProducerConfig),

       FlinkKafkaProducer.Semantic.AT_LEAST_ONCE);

Metrics

We propose to introduce the following Gauge metrics:

...

If there are alternative ways of accomplishing the same thing, what were they? The purpose of this section is to motivate why the design is the way it is and not some other way.

References