1
\$\begingroup\$

I have two processes, and for each process, I will get different GenericRecord object (this is an open source classes) and I need to validate those GenericRecord objects. This means I cannot use a single validator as I have to validate different fields for both the process.

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

If they are valid, then I will move forward otherwise I won't move forward.

Here 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.
          GenericRecord 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));
    }
  }
}

Here 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, GenericRecord record) {
    Optional<DataHolder> dataHolder = Optional.absent();
    if (isValid(processName, record))
      dataHolder = Optional.of(buildDataHolder(processName, record));
    return dataHolder;
  }

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

  private DataHolder buildDataHolder(String processName, GenericRecord 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, GenericRecord 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, GenericRecord 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, GenericRecord 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 here is my ValidatorB class in which I am validating a few different fields as compared to ValidatorA on a record object. 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, GenericRecord record) {
    Optional<DataHolder> dataHolder = Optional.absent();
    if (isValid(processName, record))
      dataHolder = Optional.of(buildDataHolder(processName, record));
    return dataHolder;
  }

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

  private DataHolder buildDataHolder(String processName, GenericRecord 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).build();
  }


  private boolean isValidType(String processName, GenericRecord 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, GenericRecord 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, GenericRecord record) {
    String itemId = (String) DataUtils.extract(record, "itemId");
    if (Strings.isNullOrEmpty(itemId)) {
      logger.logError("invalid itemId is coming.");
      return false;
    }
    return true;
  }
}

And here is my abstract class:

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

Questions:

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 arguments.

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 based 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.
  • Is the way I have my abstract class Validator right? It is not doing any useful things at all looks like.

Each of my Validator are 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(). Each processName should work with their own Validator so processA cannot work with ValidatorB.

I'd like to see if there is any better way to do this.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.