def send(record:,
             schema_name:,
             namespace: nil,
             topic: nil,
             syncronous: false)
      topic ||= schema_name
      encoded = @avro.encode(record,
                             schema_name: schema_name,
                             namespace: namespace)
      if syncronous
        @sync.produce(encoded, topic: topic)
        @sync.deliver_messages
      else
        @async.produce(encoded, topic: topic)
      end
    ensure
      @async.shutdown
      @sync.shutdown
    end
def send(record:,
             schema_name:,
             namespace: nil,
             topic: nil,
             syncronous: false)
      topic ||= schema_name
      encoded = @avro.encode(record,
                             schema_name: schema_name,
                             namespace: namespace)
      if syncronous
        @sync.produce(encoded, topic: topic)
        @sync.deliver_messages
      else
        @async.produce(encoded, topic: topic)
      end
    ensure
      @async.shutdown
      @sync.shutdown
    end
avro   - Avro instance for deserializing records [AvroTurf::Messaging]
 kafka  - Kafka instance for creating producers [Kafka]
 config - producer configuration (optional) [Hash]
 Commit a record to a topic.

 record       - record to serialize and commit [String]
 schema_name  - name of schema to encode record from [String]
 namespace    - namespace of schema (optional) [String]
 topic        - name of topic. Will be schema_name if nil (optional) [String]
 synchronous  - blocks until commit if true (optional) [Boolean]