Repository: apache/kafka
File: core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala (Scala)

Code before:
    ApiKeys.ALTER_REPLICA_LOG_DIRS -> clusterAlterAcl,
    ApiKeys.DESCRIBE_LOG_DIRS -> clusterDescribeAcl,
    ApiKeys.CREATE_PARTITIONS -> topicAlterAcl,
    ApiKeys.ELECT_LEADERS -> clusterAlterAcl,
    ApiKeys.INCREMENTAL_ALTER_CONFIGS -> topicAlterConfigsAcl,
    ApiKeys.ALTER_PARTITION_REASSIGNMENTS -> clusterAlterAcl,
    ApiKeys.LIST_PARTITION_REASSIGNMENTS -> clusterDescribeAcl,
    ApiKeys.OFFSET_DELETE -> groupReadAcl,
    ApiKeys.DESCRIBE_PRODUCERS -> topicReadAcl,
    ApiKeys.DESCRIBE_TRANSACTIONS -> transactionalIdDescribeAcl,
    ApiKeys.CONSUMER_GROUP_HEARTBEAT -> groupReadAcl,
    ApiKeys.CONSUMER_GROUP_DESCRIBE -> groupDescribeAcl,
    ApiKeys.SHARE_GROUP_HEARTBEAT -> (shareGroupReadAcl ++ topicDescribeAcl),
    ApiKeys.SHARE_GROUP_DESCRIBE -> (shareGroupDescribeAcl ++ topicDescribeAcl),
    ApiKeys.SHARE_FETCH -> (shareGroupReadAcl ++ topicReadAcl),
    ApiKeys.SHARE_ACKNOWLEDGE -> (shareGroupReadAcl ++ topicReadAcl),
    ApiKeys.INITIALIZE_SHARE_GROUP_STATE -> clusterAcl,
    ApiKeys.READ_SHARE_GROUP_STATE -> clusterAcl,
    ApiKeys.WRITE_SHARE_GROUP_STATE -> clusterAcl,
    ApiKeys.DELETE_SHARE_GROUP_STATE -> clusterAcl,
    ApiKeys.READ_SHARE_GROUP_STATE_SUMMARY -> clusterAcl,
    ApiKeys.DESCRIBE_SHARE_GROUP_OFFSETS -> (shareGroupDescribeAcl ++ topicDescribeAcl),
    ApiKeys.DELETE_SHARE_GROUP_OFFSETS -> (shareGroupDeleteAcl ++ topicReadAcl)
  )

  private def createMetadataRequest(allowAutoTopicCreation: Boolean) = {
    new requests.MetadataRequest.Builder(java.util.List.of(topic), allowAutoTopicCreation).build()
  }

  private def createProduceRequest(name: String, id: Uuid, version: Short) = {
    requests.ProduceRequest.builder(new ProduceRequestData()
        .setTopicData(new ProduceRequestData.TopicProduceDataCollection(
          util.List.of(new ProduceRequestData.TopicProduceData()
              .setName(name)
              .setTopicId(id)
              .setPartitionData(util.List.of(
                new ProduceRequestData.PartitionProduceData()
                  .setIndex(tp.partition)
                  .setRecords(MemoryRecords.withRecords(Compression.NONE, new SimpleRecord("test".getBytes))))))
            .iterator))
        .setAcks(1.toShort)
        .setTimeoutMs(5000))
      .build(version)
  }

  private def createFetchRequest = {
    val partitionMap = new util.LinkedHashMap[TopicPartition, requests.FetchRequest.PartitionData]
    partitionMap.put(tp, new requests.FetchRequest.PartitionData(getTopicIds().getOrElse(tp.topic, Uuid.ZERO_UUID),
      0, 0, 100, Optional.of(27)))
    requests.FetchRequest.Builder.forConsumer(ApiKeys.FETCH.latestVersion, 100, Int.MaxValue, partitionMap).build()
  }

Review comment (security):
I have a question: why does altering offsets require group read permission instead of alter permission? 

Code after:
    ApiKeys.DESCRIBE_ACLS -> clusterDescribeAcl,
    ApiKeys.DELETE_ACLS -> clusterAlterAcl,
    ApiKeys.ALTER_REPLICA_LOG_DIRS -> clusterAlterAcl,
    ApiKeys.DESCRIBE_LOG_DIRS -> clusterDescribeAcl,
    ApiKeys.CREATE_PARTITIONS -> topicAlterAcl,
    ApiKeys.ELECT_LEADERS -> clusterAlterAcl,
    ApiKeys.INCREMENTAL_ALTER_CONFIGS -> topicAlterConfigsAcl,
    ApiKeys.ALTER_PARTITION_REASSIGNMENTS -> clusterAlterAcl,
    ApiKeys.LIST_PARTITION_REASSIGNMENTS -> clusterDescribeAcl,
    ApiKeys.OFFSET_DELETE -> groupReadAcl,
    ApiKeys.DESCRIBE_PRODUCERS -> topicReadAcl,
    ApiKeys.DESCRIBE_TRANSACTIONS -> transactionalIdDescribeAcl,
    ApiKeys.CONSUMER_GROUP_HEARTBEAT -> groupReadAcl,
    ApiKeys.CONSUMER_GROUP_DESCRIBE -> groupDescribeAcl,
    ApiKeys.SHARE_GROUP_HEARTBEAT -> (shareGroupReadAcl ++ topicDescribeAcl),
    ApiKeys.SHARE_GROUP_DESCRIBE -> (shareGroupDescribeAcl ++ topicDescribeAcl),
    ApiKeys.SHARE_FETCH -> (shareGroupReadAcl ++ topicReadAcl),
    ApiKeys.SHARE_ACKNOWLEDGE -> (shareGroupReadAcl ++ topicReadAcl),
    ApiKeys.INITIALIZE_SHARE_GROUP_STATE -> clusterAcl,
    ApiKeys.READ_SHARE_GROUP_STATE -> clusterAcl,
    ApiKeys.WRITE_SHARE_GROUP_STATE -> clusterAcl,
    ApiKeys.DELETE_SHARE_GROUP_STATE -> clusterAcl,
    ApiKeys.READ_SHARE_GROUP_STATE_SUMMARY -> clusterAcl,
    ApiKeys.DESCRIBE_SHARE_GROUP_OFFSETS -> (shareGroupDescribeAcl ++ topicDescribeAcl),
    ApiKeys.DELETE_SHARE_GROUP_OFFSETS -> (shareGroupDeleteAcl ++ topicReadAcl),
    ApiKeys.ALTER_SHARE_GROUP_OFFSETS -> (shareGroupReadAcl ++ topicReadAcl)
  )

  private def createMetadataRequest(allowAutoTopicCreation: Boolean) = {
    new requests.MetadataRequest.Builder(java.util.List.of(topic), allowAutoTopicCreation).build()
  }

  private def createProduceRequest(name: String, id: Uuid, version: Short) = {
    requests.ProduceRequest.builder(new ProduceRequestData()
        .setTopicData(new ProduceRequestData.TopicProduceDataCollection(
          util.List.of(new ProduceRequestData.TopicProduceData()
              .setName(name)
              .setTopicId(id)
              .setPartitionData(util.List.of(
                new ProduceRequestData.PartitionProduceData()
                  .setIndex(tp.partition)
                  .setRecords(MemoryRecords.withRecords(Compression.NONE, new SimpleRecord("test".getBytes))))))
            .iterator))
        .setAcks(1.toShort)
        .setTimeoutMs(5000))
      .build(version)
  }

  private def createFetchRequest = {
    val partitionMap = new util.LinkedHashMap[TopicPartition, requests.FetchRequest.PartitionData]
    partitionMap.put(tp, new requests.FetchRequest.PartitionData(getTopicIds().getOrElse(tp.topic, Uuid.ZERO_UUID),