Local kafka in docker with transaction support

I'm trying to experiment with Kafka Streams and transactions/exactly once locally and of course the easiest way to do it is use docker-compose. My Kafka container service/container is set up as follows:

  kafka:
    image: confluentinc/cp-kafka:6.0.1
    hostname: kafka
    container_name: kafka
    ports:
      - '9092:9092'
      - '29092:29092'
    depends_on:
      - zookeeper
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:32181
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      #Otherwise clients timeout trying to get 3 replicas:      
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1

However, my examples don't work, even though they work perfectly well before I set p.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE) I get an infinite loop like so:

[kafka-producer-network-thread | favorite-color-7921dfcb-3e64-4786-9e9a-10d719736511-StreamThread-1-0_0-producer] INFO org.apache.kafka.clients.producer.internals.TransactionManager - [Producer clientId=favorite-color-7921dfcb-3e64-4786-9e9a-10d719736511-StreamThread-1-0_0-producer, transactionalId=favorite-color-0_0] Discovered transaction coordinator localhost:9092 (id: 1 rack: null)
[kafka-producer-network-thread | favorite-color-7921dfcb-3e64-4786-9e9a-10d719736511-StreamThread-1-0_0-producer] INFO org.apache.kafka.clients.producer.internals.TransactionManager - [Producer clientId=favorite-color-7921dfcb-3e64-4786-9e9a-10d719736511-StreamThread-1-0_0-producer, transactionalId=favorite-color-0_0] Discovered transaction coordinator localhost:9092 (id: 1 rack: null)
[kafka-producer-network-thread | favorite-color-7921dfcb-3e64-4786-9e9a-10d719736511-StreamThread-1-0_0-producer] INFO org.apache.kafka.clients.producer.internals.TransactionManager - [Producer clientId=favorite-color-7921dfcb-3e64-4786-9e9a-10d719736511-StreamThread-1-0_0-producer, transactionalId=favorite-color-0_0] Discovered transaction coordinator localhost:9092 (id: 1 rack: null)

EDIT: Also adding a code snippet:

package streams

import java.util.Properties

import org.apache.kafka.streams.scala.Serdes._
import org.apache.kafka.streams.scala._
import org.apache.kafka.streams.scala.kstream._
import org.apache.kafka.streams.{KafkaStreams, StreamsConfig}

import scala.concurrent.duration.{DurationInt, FiniteDuration}
import scala.jdk.DurationConverters.ScalaDurationOps

object FavoriteColor extends App {

  System.setProperty("org.slf4j.simpleLogger.logFile", "System.out")

  import org.apache.kafka.streams.scala.ImplicitConversions._

  val props: Properties = {
    val p = new Properties()
    p.put(StreamsConfig.APPLICATION_ID_CONFIG, "favorite-color")
    p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")
    p.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, "0")
    p.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE)
    p
  }

  val builder: StreamsBuilder = new StreamsBuilder
  private val inputTopic = "user-colors"
  private val intermediateTopic = "table-input-topic"
  private val outputTopic = "count-colors-output"
  builder.stream[String, String](inputTopic)
    .filter((_, value) => value.contains(","))
    .mapValues((_, v) => v.split(","))
    .map {
      case (_, Array(userName, color)) => (userName, color)
      case (_, Array(userName, _*)) => (userName, null)
    }
    .to(intermediateTopic)

  val outputTable = builder.table[String, String](intermediateTopic)
    .groupBy((_: String, word: String) => (word, word))
    .count()(Materialized.as("count-colors"))
  outputTable.toStream.to(outputTopic)

  val streams: KafkaStreams = new KafkaStreams(builder.build(), props)
  streams.start()
  println(s"Printing topology: ${streams.toString}")

  sys.ShutdownHookThread {
    streams.close(10 seconds)
  }

  implicit def toJavaDuration(d: FiniteDuration): java.time.Duration = d.toJava
}