PRODUCE(0, "Produce"),
FETCH(1, "Fetch"),
LIST_OFFSETS(2, "Offsets"),
METADATA(3, "Metadata"),
LEADER_AND_ISR(4, "LeaderAndIsr"),
STOP_REPLICA(5, "StopReplica"),
UPDATE_METADATA_KEY(6, "UpdateMetadata"),
CONTROLLED_SHUTDOWN_KEY(7, "ControlledShutdown"),
OFFSET_COMMIT(8, "OffsetCommit"),
OFFSET_FETCH(9, "OffsetFetch"),
GROUP_COORDINATOR(10, "GroupCoordinator"),
JOIN_GROUP(11, "JoinGroup"),
HEARTBEAT(12, "Heartbeat"),
LEAVE_GROUP(13, "LeaveGroup"),
SYNC_GROUP(14, "SyncGroup"),
DESCRIBE_GROUPS(15, "DescribeGroups"),

static {int maxKey = -1;for (ApiKeys key : ()) {maxKey = (maxKey, );}codeToType = new ApiKeys[maxKey + 1];for (ApiKeys key : ()) {codeToType[] = key;}MAX_API_KEY = maxKey;
}

SASL_HANDSHAKE(17, "SaslHandshake", (), ()),
API_VERSIONS(18, "ApiVersions", (), ()) {@Overridepublic Struct parseResponse(short version, ByteBuffer buffer) {// Fallback to version 0 for ApiVersions response. If a client sends an ApiVersionsRequest// using a version higher than that supported by the broker, a version 0 response is sent// to the client indicating UNSUPPORTED_VERSION.return parseResponse(version, buffer, (short) 0);}
},
CREATE_TOPICS(19, "CreateTopics", (), ()),DELETE_TOPICS(20, "DeleteTopics", (), ()),DELETE_RECORDS(21, "DeleteRecords", (), ()),
INIT_PRODUCER_ID(22, "InitProducerId", (),()),
OFFSET_FOR_LEADER_EPOCH(23, "OffsetForLeaderEpoch", false, (),()),
ADD_PARTITIONS_TO_TXN(24, "AddPartitionsToTxn", false, RecordBatch.MAGIC_VALUE_V2,AddPartitionsToTxnRequest.schemaVersions(), ()),
ADD_OFFSETS_TO_TXN(25, "AddOffsetsToTxn", false, RecordBatch.MAGIC_VALUE_V2, AddOffsetsToTxnRequest.schemaVersions(),()),
END_TXN(26, "EndTxn", false, RecordBatch.MAGIC_VALUE_V2, EndTxnRequest.schemaVersions(),()),
WRITE_TXN_MARKERS(27, "WriteTxnMarkers", true, RecordBatch.MAGIC_VALUE_V2, WriteTxnMarkersRequest.schemaVersions(),()),
TXN_OFFSET_COMMIT(28, "TxnOffsetCommit", false, RecordBatch.MAGIC_VALUE_V2, TxnOffsetCommitRequest.schemaVersions(),()),
DESCRIBE_ACLS(29, "DescribeAcls", (), ()),
CREATE_ACLS(30, "CreateAcls", (), ()),
DELETE_ACLS(31, "DeleteAcls", (), ()),
DESCRIBE_CONFIGS(32, "DescribeConfigs", (),()),
ALTER_CONFIGS(33, "AlterConfigs", (),()),
ALTER_REPLICA_LOG_DIRS(34, "AlterReplicaLogDirs", (),()),
DESCRIBE_LOG_DIRS(35, "DescribeLogDirs", (),()),
SASL_AUTHENTICATE(36, "SaslAuthenticate", (),()),
CREATE_PARTITIONS(37, "CreatePartitions", (),()),
CREATE_DELEGATION_TOKEN(38, "CreateDelegationToken", (), ()),RENEW_DELEGATION_TOKEN(39, "RenewDelegationToken", (), ()),EXPIRE_DELEGATION_TOKEN(40, "ExpireDelegationToken", (), ()),DESCRIBE_DELEGATION_TOKEN(41, "DescribeDelegationToken", (), ()),DELETE_GROUPS(42, "DeleteGroups", (), ()),ELECT_PREFERRED_LEADERS(43, "ElectPreferredLeaders", ,

override def run() {startupComplete()while(isRunning) {try {// setup any new connections that have been queued up        configureNewConnections()// register any new responses for writing        processNewResponses()// 重写了selectortry {(300)} catch {case e @ (_: IllegalStateException | _: IOException) =>error("Closing processor %s due to illegal state or IO exception".format(id))swallow(closeAll())shutdownComplete()throw e}{ receive =>try {val channel = ()val session = (new KafkaPrincipal(KafkaPrincipal.USER_TYPE, channel.principal.getName),)val req = (processor = id, connectionId = , session = session, buffer = receive.payload, startTimeMs = time.milliseconds, securityProtocol = protocol)(req)} catch {// 重点，这里没有捕获数组越界异常!导致被外围的异常捕获,后续逻辑没有处理case e @ (_: InvalidRequestException | _: SchemaException) =>// note that even though we got an exception, we can assume that  is valid. Issues with constructing a valid receive object were handled earliererror("Closing socket for " +  + " because of error", e)close(selector, )}()}{ send =>val resp = ().getOrElse {thrownew IllegalStateException(s"Send for ${} completed, but not in `inflightResponses`")}()()}{ connectionId =>val remoteHost = (connectionId).getOrElse {thrownew IllegalStateException(s"connectionId has unexpected format: $connectionId")}.remoteHost// the channel has been closed by the selector but the quotas still need to be updated          connectionQuotas.dec((remoteHost))}} catch {// We catch all the throwables here to prevent the processor thread from exiting. We do this because// letting a processor exit might cause bigger impact on the broker. Usually the exceptions thrown would// be either associated with a specific socket channel or a bad request. We just ignore the bad socket channel// or request. This behavior might need to be reviewed if we see an exception that need the entire broker to stop.case e : ControlThrowable => throw ecase e : Throwable =>error("Processor got uncaught exception.", e)}}debug("Closing selector - processor " + id)swallowError(closeAll())shutdownComplete()
｝