import org.apache.kafka.common.serialization.Deserializer

import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient
import io.confluent.kafka.serializers.KafkaAvroDeserializer


class StringAvroDeserializer : Deserializer<String> {

    private val inner: KafkaAvroDeserializer

    constructor() {
        inner = KafkaAvroDeserializer()
    }

    /**
     * For testing purposes only.
     */
    internal constructor(client: SchemaRegistryClient) {
        inner = KafkaAvroDeserializer(client)
    }

    override fun configure(deserializerConfig: Map<String, *>,
                           isDeserializerForRecordKeys: Boolean) {
        inner.configure(deserializerConfig, isDeserializerForRecordKeys)
    }

    override fun deserialize(topic: String, bytes: ByteArray): String {
        return inner.deserialize(topic, bytes) as String
    }

    override fun close() {
        inner.close()
    }

}