Add multi-topic convenience method

[keirlawson]

The equivalent of Deserializer.topic for use with Vulcan codecs.

I found myself having to write this when creating a consumer that read from multiple topics, hopefully it might be useful to others.

[codecov-io]

Codecov Report

Merging #359 into master will increase coverage by 0.04%.
The diff coverage is 100.00%.

@@            Coverage Diff             @@
##           master     #359      +/-   ##
==========================================
+ Coverage   91.29%   91.34%   +0.04%     
==========================================
  Files          62       62              
  Lines        1447     1455       +8     
  Branches       54       54              
==========================================
+ Hits         1321     1329       +8     
  Misses        126      126              

Impacted Files	Coverage Δ
...main/scala/fs2/kafka/vulcan/AvroDeserializer.scala	100.00% <100.00%> (ø)

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 66b6d52...f6a80af. Read the comment docs.

[vlovgr]

Thanks @keirlawson! Is there anything that prevents you from creating a coproduct of the events from the different topics? If you define a union Codec for such a coproduct, then you can simply use that type to deserialize records from all topics, without having to rely on topic names.

[SystemFw]

Hey @vlovgr , I had at least a use case where this was still useful, since that was some information encoded in the topic names that had to be propagated the Vulcan codecs. I wrote my own solution, which is quite a lot uglier than this PR, judge by yourself:

  def multiTopicDeserializer[F[_]: Sync, A](
      codecs: PartialFunction[String, Codec[_ <: A]]
  ): AvroSettings[F] => RecordDeserializer[F, A] = {
    def toF[B](fb: Either[AvroError, B]): F[B] =
      Sync[F].fromEither(fb.leftMap(_.throwable))

    def getCodec(topic: String) =
      Sync[F].fromOption(
        codecs.lift.apply(topic),
        new KafkaException(s"fs2.kafka.UnexpectedTopicException: unexpected topic [$topic]")
      )

    def deserializer(
        native: KafkaAvroDeserializer,
        client: SchemaRegistryClient
    ): Deserializer[F, A] =
      Deserializer
        .instance { (topic, _, bytes) =>
          getCodec(topic).flatMap { codec =>
            toF(codec.schema).flatMap { schema =>
              Sync[F].suspend {
                val writerSchemaId = ByteBuffer.wrap(bytes).getInt(1) // skip magic byte
                val writerSchema = client.getById(writerSchemaId)

                toF(codec.decode(native.deserialize(topic, bytes, schema), writerSchema))
              }
            }
          }
        }

    settings =>
      RecordDeserializer.instance(
        forKey = settings.createAvroDeserializer(true).map((deserializer _).tupled),
        forValue = settings.createAvroDeserializer(false).map((deserializer _).tupled)
      )
  }

[keirlawson]

@vlovgr happy to hear there's an easier way :-) I just tried using Codec.union to achieve this however and it doesn't seem to work, when I run I get vulcan.AvroException: Got unexpected schema type RECORD while decoding union, expected schema type UNION

[marcodippy]

I just tried using Codec.union to achieve this however and it doesn't seem to work, when I run I get vulcan.AvroException: Got unexpected schema type RECORD while decoding union, expected schema type UNION

hey @vlovgr :) I just wanted to confirm that that trick doesn't work anymore with the latest fs2-kafka / vulcan

[vlovgr]

Yes, I believe that stopped working with #282. I've created a pull request (fd4s/vulcan#191) to change the union decoding to also accept schemas in the union (in addition to the union schema), which should fix the issue.

Sounds like the addition in this pull request should still be useful in special circumstances.

[mwz]

I had at least a use case where this was still useful, since that was some information encoded in the topic names that had to be propagated the Vulcan codecs.

Hey @SystemFw, I'm quite curious about your use case, can you share any more details on why you need to do this? (feel free to DM me if you don't want to talk about it publicly)

[SystemFw]

Is there any reason why this never got merged?

[vlovgr]

I took a quick look now, and I think this assumption about forValue makes it only work for values.

[SystemFw]

does the uglier solution I posted above suffer from the same limitation?

[vlovgr]

No, that solution looks alright to me. 👍

[SystemFw]

I'll see if I can PR that then

[aartigao]

I'll see if I can PR that then

Hey @SystemFw, I can't find your PR for that. Do you feel this is still needed after +3 years? Shall we re-discuss it for 3.x?