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

Code before:
    val appendResult = client.incrementalAlterConfigs(util.Map.of(topicResource, topicAppendConfigs))
    appendResult.all.get

    // Subtract values that are not present
    val topicSubtractConfigs = util.List.of(
      new AlterConfigOp(new ConfigEntry(QuotaConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, subtractValues), AlterConfigOp.OpType.SUBTRACT)
    )
    val subtractResult = client.incrementalAlterConfigs(util.Map.of(topicResource, topicSubtractConfigs))
    subtractResult.all.get

    ensureConsistentKRaftMetadata()

    // Verify that topics were updated correctly
    val describeResult = client.describeConfigs(util.List.of(topicResource))
    val configs = describeResult.all.get

    assertEquals(appendValues, configs.get(topicResource).get(QuotaConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG).value)
  }

  @Test
  def testIncrementalAlterConfigsDeleteAndSetBrokerConfigs(): Unit = {
    client = createAdminClient
    val broker0Resource = new ConfigResource(ConfigResource.Type.BROKER, "0")
    client.incrementalAlterConfigs(util.Map.of(broker0Resource,
      util.List.of(new AlterConfigOp(new ConfigEntry(QuotaConfig.LEADER_REPLICATION_THROTTLED_RATE_CONFIG, "123"),
        AlterConfigOp.OpType.SET),
        new AlterConfigOp(new ConfigEntry(QuotaConfig.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG, "456"),
          AlterConfigOp.OpType.SET)
      ))).all().get()
    TestUtils.waitUntilTrue(() => {
      val broker0Configs = client.describeConfigs(util.List.of(broker0Resource)).
        all().get().get(broker0Resource).entries().asScala.map(entry => (entry.name, entry.value)).toMap
      "123".equals(broker0Configs.getOrElse(QuotaConfig.LEADER_REPLICATION_THROTTLED_RATE_CONFIG, "")) &&
        "456".equals(broker0Configs.getOrElse(QuotaConfig.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG, ""))
    }, "Expected to see the broker properties we just set", pause=25)
    client.incrementalAlterConfigs(util.Map.of(broker0Resource,
      util.List.of(new AlterConfigOp(new ConfigEntry(QuotaConfig.LEADER_REPLICATION_THROTTLED_RATE_CONFIG, ""),
        AlterConfigOp.OpType.DELETE),
        new AlterConfigOp(new ConfigEntry(QuotaConfig.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG, "654"),
          AlterConfigOp.OpType.SET),
        new AlterConfigOp(new ConfigEntry(QuotaConfig.REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_CONFIG, "987"),
          AlterConfigOp.OpType.SET)
      ))).all().get()
    TestUtils.waitUntilTrue(() => {
      val broker0Configs = client.describeConfigs(util.List.of(broker0Resource)).
        all().get().get(broker0Resource).entries().asScala.map(entry => (entry.name, entry.value)).toMap
      "".equals(broker0Configs.getOrElse(QuotaConfig.LEADER_REPLICATION_THROTTLED_RATE_CONFIG, "")) &&
        "654".equals(broker0Configs.getOrElse(QuotaConfig.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG, "")) &&
        "987".equals(broker0Configs.getOrElse(QuotaConfig.REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_CONFIG, ""))
    }, "Expected to see the broker properties we just modified", pause=25)
  }

Review comment (suggestion):
```suggestion
          AlterConfigOp.OpType.SET),
```

Code after:
    val appendResult = client.incrementalAlterConfigs(util.Map.of(topicResource, topicAppendConfigs))
    appendResult.all.get

    // Subtract values that are not present
    val topicSubtractConfigs = util.List.of(
      new AlterConfigOp(new ConfigEntry(QuotaConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG, subtractValues), AlterConfigOp.OpType.SUBTRACT)
    )
    val subtractResult = client.incrementalAlterConfigs(util.Map.of(topicResource, topicSubtractConfigs))
    subtractResult.all.get

    ensureConsistentKRaftMetadata()

    // Verify that topics were updated correctly
    val describeResult = client.describeConfigs(util.List.of(topicResource))
    val configs = describeResult.all.get

    assertEquals(appendValues, configs.get(topicResource).get(QuotaConfig.LEADER_REPLICATION_THROTTLED_REPLICAS_CONFIG).value)
  }

  @Test
  def testIncrementalAlterConfigsDeleteAndSetBrokerConfigs(): Unit = {
    client = createAdminClient
    val broker0Resource = new ConfigResource(ConfigResource.Type.BROKER, "0")
    client.incrementalAlterConfigs(util.Map.of(broker0Resource,
      util.List.of(new AlterConfigOp(new ConfigEntry(QuotaConfig.LEADER_REPLICATION_THROTTLED_RATE_CONFIG, "123"),
          AlterConfigOp.OpType.SET),
        new AlterConfigOp(new ConfigEntry(QuotaConfig.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG, "456"),
          AlterConfigOp.OpType.SET)
      ))).all().get()
    TestUtils.waitUntilTrue(() => {
      val broker0Configs = client.describeConfigs(util.List.of(broker0Resource)).
        all().get().get(broker0Resource).entries().asScala.map(entry => (entry.name, entry.value)).toMap
      "123".equals(broker0Configs.getOrElse(QuotaConfig.LEADER_REPLICATION_THROTTLED_RATE_CONFIG, "")) &&
        "456".equals(broker0Configs.getOrElse(QuotaConfig.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG, ""))
    }, "Expected to see the broker properties we just set", pause=25)
    client.incrementalAlterConfigs(util.Map.of(broker0Resource,
      util.List.of(new AlterConfigOp(new ConfigEntry(QuotaConfig.LEADER_REPLICATION_THROTTLED_RATE_CONFIG, ""),
        AlterConfigOp.OpType.DELETE),
        new AlterConfigOp(new ConfigEntry(QuotaConfig.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG, "654"),
          AlterConfigOp.OpType.SET),
        new AlterConfigOp(new ConfigEntry(QuotaConfig.REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_CONFIG, "987"),
          AlterConfigOp.OpType.SET)
      ))).all().get()
    TestUtils.waitUntilTrue(() => {
      val broker0Configs = client.describeConfigs(util.List.of(broker0Resource)).
        all().get().get(broker0Resource).entries().asScala.map(entry => (entry.name, entry.value)).toMap
      "".equals(broker0Configs.getOrElse(QuotaConfig.LEADER_REPLICATION_THROTTLED_RATE_CONFIG, "")) &&
        "654".equals(broker0Configs.getOrElse(QuotaConfig.FOLLOWER_REPLICATION_THROTTLED_RATE_CONFIG, "")) &&
        "987".equals(broker0Configs.getOrElse(QuotaConfig.REPLICA_ALTER_LOG_DIRS_IO_MAX_BYTES_PER_SECOND_CONFIG, ""))
    }, "Expected to see the broker properties we just modified", pause=25)
  }