class MessageReaderFactory {

  public static MessageReader getMessageReader(String messageReaderKlass) {
    if (messageReaderKlass == null) {
      throw UserException.validationError()
          .message("Please configure message reader implementation using the property 'store.kafka.record.reader'")
          .build(logger);
    }

    MessageReader messageReader = null;
    try {
      Class<?> klass = Class.forName(messageReaderKlass);
      if (MessageReader.class.isAssignableFrom(klass)) {
        messageReader = (MessageReader) klass.newInstance();
        logger.debug("Initialized Message Reader : {}", messageReader);
      }
    } catch (ClassNotFoundException | InstantiationException | IllegalAccessException e) {
      throw UserException.validationError().message("Failed to initialize message reader : %s", messageReaderKlass)
          .build(logger);
    }

    if (messageReader == null) {
      throw UserException.validationError().message("Message reader configured '%s' does not implement '%s'",
          messageReaderKlass, MessageReader.class.getName()).build(logger);
    }
    return messageReader;
  }

}