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

Code before:
                                configsToRemove: List[String] = List()): ShareConsumer[K, V] = {
    val props = new Properties
    props ++= shareConsumerConfig
    props ++= configOverrides
    configsToRemove.foreach(props.remove(_))
    val shareConsumer = new KafkaShareConsumer[K, V](props, keyDeserializer, valueDeserializer)
    shareConsumers += shareConsumer
    shareConsumer
  }

  def createStreamsConsumer[K, V](keyDeserializer: Deserializer[K] = new ByteArrayDeserializer,
                                valueDeserializer: Deserializer[V] = new ByteArrayDeserializer,
                                configOverrides: Properties = new Properties,
                                configsToRemove: List[String] = List(),
                                streamsRebalanceData: StreamsRebalanceData): AsyncKafkaConsumer[K, V] = {
    val props = new Properties
    props ++= streamsConsumerConfig
    props ++= configOverrides
    configsToRemove.foreach(props.remove(_))
    val streamsConsumer = new AsyncKafkaConsumer[K, V](
      new ConsumerConfig(ConsumerConfig.appendDeserializerToConfig(Utils.propsToMap(props), keyDeserializer, valueDeserializer)),
      keyDeserializer,
      valueDeserializer,
      Optional.of(streamsRebalanceData)
    )
    streamsConsumers += streamsConsumer
    streamsConsumer
  }

  def createStreamsGroup[K, V](configOverrides: Properties = new Properties,
                               configsToRemove: List[String] = List(),
                               inputTopic: String,
                               outputTopic: String,
                               streamsGroupId: String,
                               groupProtocol: String): KafkaStreams = {
    val streamsConfig = new Properties(streamsGroupConfig)
    streamsConfig.put(StreamsConfig.APPLICATION_ID_CONFIG, streamsGroupId)
    streamsConfig.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, groupProtocol)
    streamsConfig ++= configOverrides
    configsToRemove.foreach(streamsConfig.remove(_))
    val builder = new StreamsBuilder()
    builder.stream[K, V](inputTopic).to(outputTopic)
    val streams = new KafkaStreams(builder.build(), streamsConfig)
    streams
  }

  def createStreamsGroupWithAggregation[K, V](configOverrides: Properties = new Properties,
                                              configsToRemove: List[String] = List(),
                                              inputTopic: String,
                                              outputTopic: String,
                                              streamsGroupId: String,

Review comment (security):
I wonder if could create the streams groups without adding a dependency to the streams package - this seems to create an extra dependency that seems quite circular. It's not too bad since its a testDependency only.

Can you check AuthorizerIntegrationTest.createStreamsGroupToDescribe and check how I created a streams groups with input topic / output topic and also a changelog topic? This should work here as well.

Code after:
    configsToRemove.foreach(props.remove(_))
    val shareConsumer = new KafkaShareConsumer[K, V](props, keyDeserializer, valueDeserializer)
    shareConsumers += shareConsumer
    shareConsumer
  }

  def createStreamsConsumer[K, V](keyDeserializer: Deserializer[K] = new ByteArrayDeserializer,
                                valueDeserializer: Deserializer[V] = new ByteArrayDeserializer,
                                configOverrides: Properties = new Properties,
                                configsToRemove: List[String] = List(),
                                streamsRebalanceData: StreamsRebalanceData): AsyncKafkaConsumer[K, V] = {
    val props = new Properties
    props ++= streamsConsumerConfig
    props ++= configOverrides
    configsToRemove.foreach(props.remove(_))
    val streamsConsumer = new AsyncKafkaConsumer[K, V](
      new ConsumerConfig(ConsumerConfig.appendDeserializerToConfig(Utils.propsToMap(props), keyDeserializer, valueDeserializer)),
      keyDeserializer,
      valueDeserializer,
      Optional.of(streamsRebalanceData)
    )
    streamsConsumers += streamsConsumer
    streamsConsumer
  }

  def createStreamsGroup[K, V](configOverrides: Properties = new Properties,
                               configsToRemove: List[String] = List(),
                               inputTopic: String,
                               streamsGroupId: String): AsyncKafkaConsumer[K, V] = {
    val props = new Properties()
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers())
    props.put(ConsumerConfig.GROUP_ID_CONFIG, streamsGroupId)
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, classOf[ByteArrayDeserializer].getName)
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, classOf[ByteArrayDeserializer].getName)
    props ++= configOverrides
    configsToRemove.foreach(props.remove(_))

    val streamsRebalanceData = new StreamsRebalanceData(
      UUID.randomUUID(),
      Optional.empty(),
      util.Map.of(
        "subtopology-0", new StreamsRebalanceData.Subtopology(
          util.Set.of(inputTopic),
          util.Set.of(),
          util.Map.of(),
          util.Map.of(inputTopic + "-store-changelog", new StreamsRebalanceData.TopicInfo(Optional.of(1), Optional.empty(), util.Map.of())),
          util.Set.of()
        )),
      Map.empty[String, String].asJava
    )