code: protected ProducerCloseable<String, byte[]> createProducer(ProducerConfig config) {
      return this.closer.register(new ProducerCloseable<String, byte[]>(config));
    }
docs: Actually creates the Kafka producer.