import org.apache.kafka.common.serialization.Serde
import org.apache.kafka.streams.scala.StreamsBuilder
import org.apache.kafka.streams.scala.kstream.{Consumed, KStream}
import io.confluent.kafka.streams.serdes.avro.GenericAvroSerde

import scala.collection.JavaConverters._

object Solution {

  trait IsKeySerde
  type KeySerde[T] = Serde[T] with IsKeySerde

  trait IsValueSerde
  type ValueSerde[T] = Serde[T] with IsValueSerde

  //example of the changed implicit conversion from the org.apache.kafka.streams.scala.ImplicitConversions
  implicit def consumedFromSerdeNew[K, V](implicit keySerde: KeySerde[K], valueSerde: ValueSerde[V]): Consumed[K, V] =
    Consumed.`with`[K, V]
  
}

object OldBehavior {
  import Solution._
  implicit def asKeySerde[T](serde: Serde[T]): KeySerde[T] = serde.asInstanceOf[KeySerde[T]]
  implicit def asValueSerde[T](serde: Serde[T]): ValueSerde[T] = serde.asInstanceOf[ValueSerde[T]]
}

object SerdesSupport {
  import Solution._
  import OldBehavior._

  // I don't want to go too deep into implementation details,
  // so lets define some dummy type class called AlmostSerde and assume that,
  // if I have an instance of that type class for type T and GenericAvroSerde, I can create instance of Serde for T
  class AlmostSerde[T]()
  implicit def deriveAlmostSerde[T]: AlmostSerde[T] = ??? // assume that we can derive AlmostSerde instances for all T
  private def createSerde[T](as: AlmostSerde[T], gas: GenericAvroSerde): Serde[T] = ???

  implicit def specificSerdeForKey[T >: Null](implicit as: AlmostSerde[T]): KeySerde[T] = {
    val genericSerde = new GenericAvroSerde()
    val configuration = Map[String, Any]().asJava
    val isSerdeForKey = true
    genericSerde.configure(configuration, isSerdeForKey)
    createSerde(as, genericSerde)
  }

  implicit def specificSerdeForValue[T >: Null](implicit as: AlmostSerde[T]): ValueSerde[T] = {
    val genericSerde = new GenericAvroSerde()
    val configuration = Map[String, Any]().asJava
    val isSerdeForKey = false
    genericSerde.configure(configuration, isSerdeForKey)
    createSerde(as, genericSerde)
  }

}

object TestApp extends App {
  case class A(x: Int)
  case class B(y: String)

  import Solution._
  import SerdesSupport._

  val builder = new StreamsBuilder()

  val streamAtoB: KStream[A, B] = builder.stream("topic")
  val streamBtoA: KStream[B, A] = streamAtoB.map((a, b) => b -> a)
}