flink引出的kafka不同版本的兼容性Kafka协议兼容性改进

<dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-connector-kafka_${scala.binary.version}artifactId>
            <version>${flink.version}version>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.kafkagroupId>
                    <artifactId>kafka-clentsartifactId>
                exclusion>
            exclusions>
            <scope>providedscope>
        dependency>
        <dependency>
            <groupId>org.apache.kafkagroupId>
            <artifactId>kafka-clentsartifactId>
            <version>1.0.0version>
        dependency>

Fetch API (Key: 1):
Requests:
Fetch Request (Version: 0) => replica_id max_wait_time min_bytes [topics] 
  replica_id => INT32
  max_wait_time => INT32
  min_bytes => INT32
  topics => topic [partitions] 
    topic => STRING
    partitiOns=> partition fetch_offset partition_max_bytes 
      partition => INT32
      fetch_offset => INT64
      partition_max_bytes => INT32

Fetch Request (Version: 1) => replica_id max_wait_time min_bytes [topics] 
  replica_id => INT32
  max_wait_time => INT32
  min_bytes => INT32
  topics => topic [partitions] 
    topic => STRING
    partitiOns=> partition fetch_offset partition_max_bytes 
      partition => INT32
      fetch_offset => INT64
      partition_max_bytes => INT32


Responses:
Fetch Response (Version: 0) => [responses] 
  respOnses=> topic [partition_responses] 
    topic => STRING
    partition_respOnses=> partition_header record_set 
      partition_header => partition error_code high_watermark 
        partition => INT32
        error_code => INT16
        high_watermark => INT64
      record_set => RECORDS
Fetch Response (Version: 1) => throttle_time_ms [responses] 
  throttle_time_ms => INT32
  respOnses=> topic [partition_responses] 
    topic => STRING
    partition_respOnses=> partition_header record_set 
      partition_header => partition error_code high_watermark 
        partition => INT32
        error_code => INT16
        high_watermark => INT64
      record_set => RECORDS

./bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092 （我的kafka版本是：kafka_2.11-2.2.0）
localhost:9092 (id: 0 rack: null) -> (
    Produce(0): 0 to 7 [usable: 7],
    Fetch(1): 0 to 10 [usable: 10],
    ListOffsets(2): 0 to 5 [usable: 5],
    Metadata(3): 0 to 7 [usable: 7],
    LeaderAndIsr(4): 0 to 2 [usable: 2],
    StopReplica(5): 0 to 1 [usable: 1],
    UpdateMetadata(6): 0 to 5 [usable: 5],
    ControlledShutdown(7): 0 to 2 [usable: 2],
    OffsetCommit(8): 0 to 6 [usable: 6],
    OffsetFetch(9): 0 to 5 [usable: 5],
    FindCoordinator(10): 0 to 2 [usable: 2],
    JoinGroup(11): 0 to 4 [usable: 4],
    Heartbeat(12): 0 to 2 [usable: 2],
    LeaveGroup(13): 0 to 2 [usable: 2],
    SyncGroup(14): 0 to 2 [usable: 2],
    DescribeGroups(15): 0 to 2 [usable: 2],
    ListGroups(16): 0 to 2 [usable: 2],
    SaslHandshake(17): 0 to 1 [usable: 1],
    ApiVersions(18): 0 to 2 [usable: 2],
    CreateTopics(19): 0 to 3 [usable: 3],
    DeleteTopics(20): 0 to 3 [usable: 3],
    DeleteRecords(21): 0 to 1 [usable: 1],
    InitProducerId(22): 0 to 1 [usable: 1],
    OffsetForLeaderEpoch(23): 0 to 2 [usable: 2],
    AddPartitionsToTxn(24): 0 to 1 [usable: 1],
    AddOffsetsToTxn(25): 0 to 1 [usable: 1],
    EndTxn(26): 0 to 1 [usable: 1],
    WriteTxnMarkers(27): 0 [usable: 0],
    TxnOffsetCommit(28): 0 to 2 [usable: 2],
    DescribeAcls(29): 0 to 1 [usable: 1],
    CreateAcls(30): 0 to 1 [usable: 1],
    DeleteAcls(31): 0 to 1 [usable: 1],
    DescribeConfigs(32): 0 to 2 [usable: 2],
    AlterConfigs(33): 0 to 1 [usable: 1],
    AlterReplicaLogDirs(34): 0 to 1 [usable: 1],
    DescribeLogDirs(35): 0 to 1 [usable: 1],
    SaslAuthenticate(36): 0 to 1 [usable: 1],
    CreatePartitions(37): 0 to 1 [usable: 1],
    CreateDelegationToken(38): 0 to 1 [usable: 1],
    RenewDelegationToken(39): 0 to 1 [usable: 1],
    ExpireDelegationToken(40): 0 to 1 [usable: 1],
    DescribeDelegationToken(41): 0 to 1 [usable: 1],
    DeleteGroups(42): 0 to 1 [usable: 1],
    ElectPreferredLeaders(43): 0 [usable: 0]
)

public static Schema[] schemaVersions() {
        return new Schema[]{FETCH_REQUEST_V0, FETCH_REQUEST_V1, FETCH_REQUEST_V2, FETCH_REQUEST_V3, FETCH_REQUEST_V4,
            FETCH_REQUEST_V5, FETCH_REQUEST_V6, FETCH_REQUEST_V7, FETCH_REQUEST_V8};
    };

public static Schema[] schemaVersions() {
        return new Schema[] {FETCH_RESPONSE_V0, FETCH_RESPONSE_V1, FETCH_RESPONSE_V2,
            FETCH_RESPONSE_V3, FETCH_RESPONSE_V4, FETCH_RESPONSE_V5, FETCH_RESPONSE_V6,
            FETCH_RESPONSE_V7, FETCH_RESPONSE_V8};
    }