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

Code before:
import org.apache.kafka.raft.MetadataLogConfig
import org.apache.kafka.server.config.{KRaftConfigs, ReplicationConfigs}

import org.junit.jupiter.api.{AfterEach, BeforeEach, TestInfo}

import scala.collection.mutable
import scala.collection.Seq
import scala.jdk.CollectionConverters._
import scala.jdk.javaapi.OptionConverters

/**
 * A helper class for writing integration tests that involve producers, consumers, and servers
 */
abstract class IntegrationTestHarness extends KafkaServerTestHarness {
  protected def brokerCount: Int
  protected def logDirCount: Int = 1

  val producerConfig = new Properties
  val consumerConfig = new Properties
  val shareConsumerConfig = new Properties
  val streamsConsumerConfig = new Properties
  val adminClientConfig = new Properties
  val superuserClientConfig = new Properties
  val serverConfig = new Properties
  val controllerConfig = new Properties
  var streamsGroupConfig = new Properties

  private val consumers = mutable.Buffer[Consumer[_, _]]()
  private val shareConsumers = mutable.Buffer[ShareConsumer[_, _]]()
  private val streamsConsumers = mutable.Buffer[Consumer[_, _]]()
  private val producers = mutable.Buffer[KafkaProducer[_, _]]()
  private val adminClients = mutable.Buffer[Admin]()

  protected def interBrokerListenerName: ListenerName = listenerName

  protected def modifyConfigs(props: Seq[Properties]): Unit = {
    props.foreach(_ ++= serverConfig)
  }

  override def generateConfigs: Seq[KafkaConfig] = {
    val cfgs = TestUtils.createBrokerConfigs(brokerCount, interBrokerSecurityProtocol = Some(securityProtocol),
      trustStoreFile = trustStoreFile, saslProperties = serverSaslProperties, logDirCount = logDirCount)
    configureListeners(cfgs)
    modifyConfigs(cfgs)
    cfgs.foreach(_.setProperty(MetadataLogConfig.METADATA_LOG_DIR_CONFIG, TestUtils.tempDir().getAbsolutePath))
    insertControllerListenersIfNeeded(cfgs)
    cfgs.map(KafkaConfig.fromProps)
  }

  override protected def kraftControllerConfigs(testInfo: TestInfo): Seq[Properties] = {
    Seq(controllerConfig)

Review comment (suggestion):
The `streamsGroupConfig` variable is declared but never used in the code. Consider removing it or documenting its intended purpose.
```suggestion

```

Code after:
import org.apache.kafka.raft.MetadataLogConfig
import org.apache.kafka.server.config.{KRaftConfigs, ReplicationConfigs}
import org.junit.jupiter.api.{AfterEach, BeforeEach, TestInfo}

import scala.collection.mutable
import scala.collection.Seq
import scala.jdk.CollectionConverters._
import scala.jdk.javaapi.OptionConverters

/**
 * A helper class for writing integration tests that involve producers, consumers, and servers
 */
abstract class IntegrationTestHarness extends KafkaServerTestHarness {
  protected def brokerCount: Int
  protected def logDirCount: Int = 1

  val producerConfig = new Properties
  val consumerConfig = new Properties
  val shareConsumerConfig = new Properties
  val streamsConsumerConfig = new Properties
  val adminClientConfig = new Properties
  val superuserClientConfig = new Properties
  val serverConfig = new Properties
  val controllerConfig = new Properties

  private val consumers = mutable.Buffer[Consumer[_, _]]()
  private val shareConsumers = mutable.Buffer[ShareConsumer[_, _]]()
  private val streamsConsumers = mutable.Buffer[Consumer[_, _]]()
  private val producers = mutable.Buffer[KafkaProducer[_, _]]()
  private val adminClients = mutable.Buffer[Admin]()

  protected def interBrokerListenerName: ListenerName = listenerName

  protected def modifyConfigs(props: Seq[Properties]): Unit = {
    props.foreach(_ ++= serverConfig)
  }

  override def generateConfigs: Seq[KafkaConfig] = {
    val cfgs = TestUtils.createBrokerConfigs(brokerCount, interBrokerSecurityProtocol = Some(securityProtocol),
      trustStoreFile = trustStoreFile, saslProperties = serverSaslProperties, logDirCount = logDirCount)
    configureListeners(cfgs)
    modifyConfigs(cfgs)
    cfgs.foreach(_.setProperty(MetadataLogConfig.METADATA_LOG_DIR_CONFIG, TestUtils.tempDir().getAbsolutePath))
    insertControllerListenersIfNeeded(cfgs)
    cfgs.map(KafkaConfig.fromProps)
  }

  override protected def kraftControllerConfigs(testInfo: TestInfo): Seq[Properties] = {
    Seq(controllerConfig)
  }