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

Code before:
      nonExistentTp1, validAssignment,
      nonExistentTp2, validAssignment
    )).values()
    assertFutureThrows(classOf[UnknownTopicOrPartitionException], nonExistentPartitionsResult.get(nonExistentTp1))
    assertFutureThrows(classOf[UnknownTopicOrPartitionException], nonExistentPartitionsResult.get(nonExistentTp2))

    val extraNonExistentReplica = Optional.of(new NewPartitionReassignment((0 until brokerCount + 1).map(_.asInstanceOf[Integer]).asJava))
    val negativeIdReplica = Optional.of(new NewPartitionReassignment(Seq(-3, -2, -1).map(_.asInstanceOf[Integer]).asJava))
    val duplicateReplica = Optional.of(new NewPartitionReassignment(Seq(0, 1, 1).map(_.asInstanceOf[Integer]).asJava))
    val invalidReplicaResult = client.alterPartitionReassignments(util.Map.of(
      tp1, extraNonExistentReplica,
      tp2, negativeIdReplica,
      tp3, duplicateReplica
    )).values()
    assertFutureThrows(classOf[InvalidReplicaAssignmentException], invalidReplicaResult.get(tp1))
    assertFutureThrows(classOf[InvalidReplicaAssignmentException], invalidReplicaResult.get(tp2))
    assertFutureThrows(classOf[InvalidReplicaAssignmentException], invalidReplicaResult.get(tp3))
  }

  @Test
  def testLongTopicNames(): Unit = {
    val client = createAdminClient
    val longTopicName = String.join("", Collections.nCopies(249, "x"))
    val invalidTopicName = String.join("", Collections.nCopies(250, "x"))
    val newTopics2 = util.List.of(new NewTopic(invalidTopicName, 3, 3.toShort),
      new NewTopic(longTopicName, 3, 3.toShort))
    val results = client.createTopics(newTopics2).values()
    assertTrue(results.containsKey(longTopicName))
    results.get(longTopicName).get()
    assertTrue(results.containsKey(invalidTopicName))
    assertFutureThrows(classOf[InvalidTopicException], results.get(invalidTopicName))
    assertFutureThrows(classOf[InvalidTopicException],
      client.alterReplicaLogDirs(
        util.Map.of(new TopicPartitionReplica(longTopicName, 0, 0), brokers(0).config.logDirs.get(0))).all())
    client.close()
  }

  // Verify that createTopics and alterConfigs fail with null values
  @Test
  def testNullConfigs(): Unit = {

    def validateLogConfig(compressionType: String): Unit = {
      ensureConsistentKRaftMetadata()
      val topicProps = brokers.head.metadataCache.topicConfig(topic)
      val logConfig = LogConfig.fromProps(util.Map.of[String, AnyRef], topicProps)

      assertEquals(compressionType, logConfig.originals.get(TopicConfig.COMPRESSION_TYPE_CONFIG))
      assertNull(logConfig.originals.get(TopicConfig.RETENTION_BYTES_CONFIG))
      assertEquals(ServerLogConfigs.LOG_RETENTION_BYTES_DEFAULT, logConfig.retentionSize)
    }


Review comment (suggestion):
```suggestion
                         new NewTopic(longTopicName, 3, 3.toShort))
```

Code after:
      nonExistentTp1, validAssignment,
      nonExistentTp2, validAssignment
    )).values()
    assertFutureThrows(classOf[UnknownTopicOrPartitionException], nonExistentPartitionsResult.get(nonExistentTp1))
    assertFutureThrows(classOf[UnknownTopicOrPartitionException], nonExistentPartitionsResult.get(nonExistentTp2))

    val extraNonExistentReplica = Optional.of(new NewPartitionReassignment((0 until brokerCount + 1).map(_.asInstanceOf[Integer]).asJava))
    val negativeIdReplica = Optional.of(new NewPartitionReassignment(Seq(-3, -2, -1).map(_.asInstanceOf[Integer]).asJava))
    val duplicateReplica = Optional.of(new NewPartitionReassignment(Seq(0, 1, 1).map(_.asInstanceOf[Integer]).asJava))
    val invalidReplicaResult = client.alterPartitionReassignments(util.Map.of(
      tp1, extraNonExistentReplica,
      tp2, negativeIdReplica,
      tp3, duplicateReplica
    )).values()
    assertFutureThrows(classOf[InvalidReplicaAssignmentException], invalidReplicaResult.get(tp1))
    assertFutureThrows(classOf[InvalidReplicaAssignmentException], invalidReplicaResult.get(tp2))
    assertFutureThrows(classOf[InvalidReplicaAssignmentException], invalidReplicaResult.get(tp3))
  }

  @Test
  def testLongTopicNames(): Unit = {
    val client = createAdminClient
    val longTopicName = String.join("", Collections.nCopies(249, "x"))
    val invalidTopicName = String.join("", Collections.nCopies(250, "x"))
    val newTopics2 = util.List.of(new NewTopic(invalidTopicName, 3, 3.toShort),
                         new NewTopic(longTopicName, 3, 3.toShort))
    val results = client.createTopics(newTopics2).values()
    assertTrue(results.containsKey(longTopicName))
    results.get(longTopicName).get()
    assertTrue(results.containsKey(invalidTopicName))
    assertFutureThrows(classOf[InvalidTopicException], results.get(invalidTopicName))
    assertFutureThrows(classOf[InvalidTopicException],
      client.alterReplicaLogDirs(
        util.Map.of(new TopicPartitionReplica(longTopicName, 0, 0), brokers(0).config.logDirs.get(0))).all())
    client.close()
  }

  // Verify that createTopics and alterConfigs fail with null values
  @Test
  def testNullConfigs(): Unit = {

    def validateLogConfig(compressionType: String): Unit = {
      ensureConsistentKRaftMetadata()
      val topicProps = brokers.head.metadataCache.topicConfig(topic)
      val logConfig = LogConfig.fromProps(util.Map.of[String, AnyRef], topicProps)

      assertEquals(compressionType, logConfig.originals.get(TopicConfig.COMPRESSION_TYPE_CONFIG))
      assertNull(logConfig.originals.get(TopicConfig.RETENTION_BYTES_CONFIG))
      assertEquals(ServerLogConfigs.LOG_RETENTION_BYTES_DEFAULT, logConfig.retentionSize)
    }