Flink - KafkaTable 解析

this.kafkaFetcher = createFetcher(
    ...
if (discoveryIntervalMillis == PARTITION_DISCOVERY_DISABLED) {
    kafkaFetcher.runFetchLoop();
} else {
    runWithPartitionDiscovery();
}

public class KafkaConsumerThread extends Thread {
    ...
    /** The handover of data and exceptions between the consumer thread and the task thread. */
    private final Handover handover;
    ...
    /** Reference to the Kafka consumer, once it is created. */
    private volatile KafkaConsumer<byte[], byte[]> consumer;

    @Override
    public void run() {
        ...
        // this is the means to talk to FlinkKafkaConsumer's main thread
        final Handover handover = this.handover;

        // This method initializes the KafkaConsumer and guarantees it is torn down properly.
        // This is important, because the consumer has multi-threading issues,
        // including concurrent 'close()' calls.
        try {
            this.consumer = getConsumer(kafkaProperties);
    ...
    }

    @VisibleForTesting
    KafkaConsumer<byte[], byte[]> getConsumer(Properties kafkaProperties) {
        return new KafkaConsumer<>(kafkaProperties);
    }

// 从 handover poll 数据
Kafka09Fetcher.runFetchLoop
            ...
            while (running) {
                // this blocks until we get the next records
                // it automatically re-throws exceptions encountered in the consumer thread
                final ConsumerRecords<byte[], byte[]> records = handover.pollNext();

// 向 handover push 数据
KafkaConsumerThread.run
                ...
                try {
                    handover.produce(records);
                    records = null;
                }

    private Properties getKafkaProperties(DescriptorProperties descriptorProperties) {
        final Properties kafkaProperties = new Properties();
        final List<Map<String, String>> propsList = descriptorProperties.getFixedIndexedProperties(
            CONNECTOR_PROPERTIES,
            Arrays.asList(CONNECTOR_PROPERTIES_KEY, CONNECTOR_PROPERTIES_VALUE));
        propsList.forEach(kv -> kafkaProperties.put(
            descriptorProperties.getString(kv.get(CONNECTOR_PROPERTIES_KEY)),
            descriptorProperties.getString(kv.get(CONNECTOR_PROPERTIES_VALUE))
        ));
        return kafkaProperties;
    }

    protected FlinkKafkaConsumerBase<Row> getKafkaConsumer(
            String topic,
            Properties properties,
            DeserializationSchema<Row> deserializationSchema) {
        FlinkKafkaConsumerBase<Row> kafkaConsumer =
                createKafkaConsumer(topic, properties, deserializationSchema);
        switch (startupMode) {
            case EARLIEST:
                kafkaConsumer.setStartFromEarliest();
                break;
            case LATEST:
                kafkaConsumer.setStartFromLatest();
                break;
            case GROUP_OFFSETS:
                kafkaConsumer.setStartFromGroupOffsets();
                break;
            case SPECIFIC_OFFSETS:
                kafkaConsumer.setStartFromSpecificOffsets(specificStartupOffsets);
                break;
        }
        return kafkaConsumer;
    }

public final class ConsumerRecord<K, V> {
    private final String topic;
    private final int partition;
    private final long offset;
    private final K key;
    private final V value;
    ...

public final class ProducerRecord<K, V> {

    private final String topic;
    private final Integer partition;
    private final K key;
    private final V value;
    ...

    public FlinkKafkaConsumer011(List<String> topics, DeserializationSchema<T> deserializer, Properties props) {
        this(topics, new KafkaDeserializationSchemaWrapper<>(deserializer), props);
    }

            while (running) {
                // this blocks until we get the next records
                // it automatically re-throws exceptions encountered in the consumer thread
                final ConsumerRecords<byte[], byte[]> records = handover.pollNext();

                // get the records for each topic partition
                for (KafkaTopicPartitionState<TopicPartition> partition : subscribedPartitionStates()) {
                    List<ConsumerRecord<byte[], byte[]>> partitionRecords =
                            records.records(partition.getKafkaPartitionHandle());
                    for (ConsumerRecord<byte[], byte[]> record : partitionRecords) {
                        final T value = deserializer.deserialize(record);

public class KafkaDeserializationSchemaWrapper<T> implements KafkaDeserializationSchema<T> {
    ...
    @Override
    public T deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
        return deserializationSchema.deserialize(record.value());
    }

    protected SinkFunction<Row> createKafkaProducer(
            String topic,
            Properties properties,
            SerializationSchema<Row> serializationSchema,
            Optional<FlinkKafkaPartitioner<Row>> partitioner) {
        return new FlinkKafkaProducer011<>(
            topic,
            new KeyedSerializationSchemaWrapper<>(serializationSchema),
            properties,
            partitioner);
    }

public class KeyedSerializationSchemaWrapper<T> implements KeyedSerializationSchema<T> {
    ...
    @Override
    public byte[] serializeKey(T element) {
        return null;
    }

    @Override
    public byte[] serializeValue(T element) {
        return serializationSchema.serialize(element);
    }
}

    public Row deserialize(byte[] message) throws IOException {
        try {
            final JsonNode root = objectMapper.readTree(message);
            return (Row) runtimeConverter.convert(objectMapper, root);
        } catch (Throwable t) {
            throw new IOException("Failed to deserialize JSON object.", t);
        }
    }

    @Override
    public int partition(T record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
        Preconditions.checkArgument(
            partitions != null && partitions.length > 0,
            "Partitions of the target topic is empty.");

        return partitions[parallelInstanceId % partitions.length];
    }

/**
 * The default partitioning strategy:
 * <ul>
 * <li>If a partition is specified in the record, use it
 * <li>If no partition is specified but a key is present choose a partition based on a hash of the key
 * <li>If no partition or key is present choose a partition in a round-robin fashion
 */
public class DefaultPartitioner implements Partitioner 
    ...