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

Code before:
    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(outputTopic + "-store-changelog", new StreamsRebalanceData.TopicInfo(Optional.of(1), Optional.empty(), util.Map.of())),
          util.Set.of()
        )),
      Map.empty[String, String].asJava
    )

    val consumer = createStreamsConsumer(
      keyDeserializer = new ByteArrayDeserializer().asInstanceOf[Deserializer[K]],
      valueDeserializer = new ByteArrayDeserializer().asInstanceOf[Deserializer[V]],
      configOverrides = props,
      streamsRebalanceData = streamsRebalanceData
    )
    consumer.subscribe(util.Set.of(inputTopic, outputTopic),
      new StreamsRebalanceListener {
        override def onTasksRevoked(tasks: util.Set[StreamsRebalanceData.TaskId]): Optional[Exception] =
          Optional.empty()
        override def onTasksAssigned(assignment: StreamsRebalanceData.Assignment): Optional[Exception] = {
          Optional.empty()
        }
        override def onAllTasksLost(): Optional[Exception] =
          Optional.empty()
      })
    consumer
  }

  def createAdminClient(
                         listenerName: ListenerName = listenerName,
                         configOverrides: Properties = new Properties
                       ): Admin = {
    val props = new Properties
    props ++= adminClientConfig
    props ++= configOverrides
    val admin = TestUtils.createAdminClient(brokers, listenerName, props)
    adminClients += admin
    admin
  }

  def createSuperuserAdminClient(

Review comment (suggestion):
Kafka streams only subscribes to the input topic, not the ouput topics. You can remove the output topic from this function altogether.

Code after:
    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
    )

    val consumer = createStreamsConsumer(
      keyDeserializer = new ByteArrayDeserializer().asInstanceOf[Deserializer[K]],
      valueDeserializer = new ByteArrayDeserializer().asInstanceOf[Deserializer[V]],
      configOverrides = props,
      streamsRebalanceData = streamsRebalanceData
    )
    consumer.subscribe(util.Set.of(inputTopic),
      new StreamsRebalanceListener {
        override def onTasksRevoked(tasks: util.Set[StreamsRebalanceData.TaskId]): Optional[Exception] =
          Optional.empty()
        override def onTasksAssigned(assignment: StreamsRebalanceData.Assignment): Optional[Exception] = {
          Optional.empty()
        }
        override def onAllTasksLost(): Optional[Exception] =
          Optional.empty()
      })
    consumer
  }

  def createAdminClient(
    listenerName: ListenerName = listenerName,
    configOverrides: Properties = new Properties
  ): Admin = {
    val props = new Properties
    props ++= adminClientConfig
    props ++= configOverrides
    val admin = TestUtils.createAdminClient(brokers, listenerName, props)
    adminClients += admin
    admin
  }

  def createSuperuserAdminClient(
    listenerName: ListenerName = listenerName,
    configOverrides: Properties = new Properties
  ): Admin = {