}
	msg := sarama.ProducerMessage{
		Topic: h.kconfig.Topic,
		Value: sarama.ByteEncoder(logJSON),
	}
	_, _, err = h.producer.SendMessage(&msg)
	if err != nil {
		atomic.StoreInt32(&h.status, statusOffline)
	} else {
		atomic.StoreInt32(&h.status, statusOnline)
	}
	return err
}

// Init initialize kafka target
func (h *Target) init() error {
	if os.Getenv("_MINIO_KAFKA_DEBUG") != "" {

			n, err := rr.ReadAt(p.buf[bufIdx], p.offset)
			if err != nil {
				switch {
				case errors.Is(err, errFileNotFound):
					atomic.StoreInt32(&missingPartsHeal, 1)
				case errors.Is(err, errFileCorrupt):
					atomic.StoreInt32(&bitrotHeal, 1)
				case errors.Is(err, errDiskNotFound):
					atomic.AddInt32(&disksNotFound, 1)
				}

				// This will be communicated upstream.