How to validate each process object from its own validator?

I have two process and for each process, I will get different Record object and I need to validate those Record object. This means I cannot use single validator as I have to validate different fields for both the process.

• For processA, I am using ValidatorA class to validate its Record object.
• For processB, I am using ValidatorB class to validate its Record object.

If they are valid, then I will move forward otherwise I won't move forward. Below is my process code both for A and B.

public class ProcessConsumer implements Runnable {
  private static final Logger logger = Logger.getInstance(ProcessConsumer.class);
  private final String processName;
  private final Validator validator;
  private final RecordProcessor<byte[], byte[]> process;

  public ProcessConsumer(String processName, Validator validator) {
    this.processName = processName;
    this.validator = validator;
    this.process = new RecordProcessor<>();
  }

  @Override
  public void run() {
    try {
      process.subscribe(getTopicsBasedOnProcessName(processName));
      ....

      while (true) {
        ConsumerRecords<byte[], byte[]> crs = process.poll(2000);
        for (ConsumerRecord<byte[], byte[]> cr : crs) {
          // record object will be different for my both the processes.
          Record record = decoder.decode(cr.value());
          Optional<DataHolder> validatedHolder = validator.getDataHolder(processName, record);
          if (!validatedHolder.isPresent()) {
            logger.logError("records dropped. payload= ", record);
            continue;
          }
          // send validatedHolder to processor class
          Processor.getInstance().execute(validatedHolder);
        }
      }
    } catch (Exception ex) {
      logger.logError("error= ", ExceptionUtils.getStackTrace(ex));
    }
  }
}

Below is my ValidatorA class in which I am validating few fields on record object and if it is valid, then I am returning DataHolder.

public class ValidatorA extends Validator {
  private static final Logger logger = Logger.getInstance(ValidatorA.class);

  @Override
  public static Optional<DataHolder> getDataHolder(String processName, Record record) {
    Optional<DataHolder> dataHolder = Optional.absent();
    if (isValid(processName, record))
      dataHolder = Optional.of(buildDataHolder(processName, record));
    return dataHolder;
  }

  private DataHolder isValid(String processName, Record record) {
    return isValidClientIdDeviceId(processName, record) && isValidPayId(processName, record)
        && isValidHolder(processName, record)
  }  

  private DataHolder buildDataHolder(String processName, Record record) {
    Map<String, String> holder = (Map<String, String>) DataUtils.extract(record, "holder");
    String deviceId = (String) DataUtils.extract(record, "deviceId");
    Integer payId = (Integer) DataUtils.extract(record, "payId");
    String clientId = (String) DataUtils.extract(record, "clientId");

    // add mandatory fields in the holder map after getting other fields
    holder.put("isClientId", (clientId == null) ? "false" : "true");
    holder.put("isDeviceId", (clientId == null) ? "true" : "false");
    holder.put("abc", (clientId == null) ? deviceId : clientId);

    return new DataHolder.Builder(record).setClientId(clientId).setDeviceId(deviceId)
        .setPayId(String.valueOf(payId)).setHolder(holder).build();
  }

  private boolean isValidHolder(String processName, Record record) {
    Map<String, String> holder = (Map<String, String>) DataUtils.extract(record, "holder");
    if (MapUtils.isEmpty(holder)) {
      logger.logError("invalid holder is coming.");
      return false;
    }
    return true;
  }

  private boolean isValidpayId(String processName, Record record) {
    Integer payId = (Integer) DataUtils.extract(record, "payId");
    if (payId == null) {
      logger.logError("invalid payId is coming.");
      return false;
    }
    return true;
  }

  private boolean isValidClientIdDeviceId(String processName, Record record) {
    String deviceId = (String) DataUtils.extract(record, "deviceId");
    String clientId = (String) DataUtils.extract(record, "clientId");
    if (Strings.isNullOrEmpty(clientId) && Strings.isNullOrEmpty(deviceId)) {
      logger.logError("invalid clientId and deviceId is coming.");
      return false;
    }
    return true;
  }
}

And below is my ValidatorB class in which I am validating few different fields as compared to ValidatorA on record object and if it is valid, then I am returning DataHolder.

public class ValidatorB extends Validator {
  private static final Logger logger = Logger.getInstance(ValidatorB.class);

  @Override
  public static Optional<DataHolder> getDataHolder(String processName, Record record) {
    Optional<DataHolder> dataHolder = Optional.absent();
    if (isValid(processName, record))
      dataHolder = Optional.of(buildDataHolder(processName, record));
    return dataHolder;
  }

  private DataHolder isValid(String processName, Record record) {
    return isValidType(processName, record) && isValidDatumId(processName, record) && isValidItemId(processName, record);
  }  

  private DataHolder buildDataHolder(String processName, Record record) {
    String type = (String) DataUtils.extract(record, "type");
    String datumId = (String) DataUtils.extract(record, "datumId");
    String itemId = (String) DataUtils.extract(record, "itemId");

    return new DataHolder.Builder(record).setType(type).setDatumId(datumId)
        .setItemId(itemId).setHolder(holder).build();
  }


  private boolean isValidType(String processName, Record record) {
    String type = (String) DataUtils.extract(record, "type");
    if (Strings.isNullOrEmpty(type)) {
      logger.logError("invalid type is coming.");
      return false;
    }
    return true;
  }  

  private boolean isValidDatumId(String processName, Record record) {
    String datumId = (String) DataUtils.extract(record, "datumId");
    if (Strings.isNullOrEmpty(datumId)) {
      logger.logError("invalid datumId is coming.");
      return false;
    }
    return true;
  }   

  private boolean isValidItemId(String processName, Record record) {
    String itemId = (String) DataUtils.extract(record, "itemId");
    if (Strings.isNullOrEmpty(itemId)) {
      logger.logError("invalid itemId is coming.");
      return false;
    }
    return true;
  }
}

And below is my abstract class:

public abstract class Validator {
  public abstract Optional<DataHolder> getDataHolder(String processName, Record record);
}

Question:

This is how I am calling for both of my process. As you can see, I am passing processName and its particular validator in the constructor argumnets.

ProcessConsumer processA = new ProcessConsumer("processA", new ValidatorA());
ProcessConsumer processB = new ProcessConsumer("processB", new ValidatorB());

• Is this a good design where for each of my process, pass its validator along with?
• Is there any way we can avoid passing that? And internally figure out what validators to use basis on the processName? I already have an enum with all my processName. I need to make this design extensible so that if I add new process in future, it should be scalable.
• Also the way I have my abstract class Validator is right? It is not doing any useful things at all looks like.

Each of my Validator is basically trying to validate whether the record object is valid or not. If they are valid, then they make DataHolder builder and return it, otherwise it returns Optional.absent();