Validate the inputs and perform actions accordingly
I think you have some great questions here! 

You asked: 


  Is there any better way to do the above things so that the code looks clean?


I'm not sure that's your only consideration here because it looks like this code is designed to be as efficient as possible. I say that because of the fact that the validation logic skips over the record as soon as a problem is found and the packetBuilder is only created after the initial validation - thereby avoiding the cost of creating the object every time through the loop regardless of the validity of the data. 

So, with all of this in mind I'm afraid that my answer to your question will have to be: there might be slight improvements you can make but you can't avoid the null checking unless the Exception thrown by the builder is very detailed about what exactly went wrong and you can use it to gather your metrics.

If you want to make the code more modular you could split the validation logic into separate methods - perhaps one for each field. I'm not sure exactly how efficient you're trying to be and there is some cost involved with calling methods, so this might not be a good change for you. However, if peak performance is required then I'm sure you're familiar with profiling so just use whatever tool you typically use and determine whether the change impacts performance.

Along the same line of thinking you might even create an object to hold the fields and provide methods for validating each of them as well as a method to validate the data in general (this method could call each of the field validation methods and determine a result based on the results for the fields). Note that this approach also has overhead cost as you would create this object each time through the loop.

Here is what the class to hold the fields might look like: 

public class PacketValidation {
    private String clientId, deviceId, payId, crossId;
    private Map&lt;String,String&gt; appPayload;
    // Not sure how you initialize the logger, so just added it as a field.
    private Logger logger;

    public PacketValidation(GenericRecord payload){
        clientId = (String) DataUtils.parseRecord(payload, "clientId");
        deviceId = (String) DataUtils.parseRecord(payload, "deviceId");
        payId = (String) DataUtils.parseRecord(payload, "payId");
        crossId = (String) DataUtils.parseRecord(payload, "crossId");
        appPayload = (Map&lt;String, String&gt;) DataUtils.parseRecord(payload, "appPayload");
    }

    public boolean isValid(){
        return isValidClientIdDeviceId() &amp;&amp; isValidPayId() &amp;&amp; isValidAppPayload();
    }

    private boolean isValidAppPayload() {
        if (MapUtils.isEmpty(appPayload)) {
            DataMetricHolder.getInstance().increment(DataUtils.EVENT_A,
                MetricName.INVALID_PAYLOAD);
            logger.logError("dropping records. invalid payload is coming.");
            return false;
          }
        return true;
    }

    private boolean isValidPayId() {
        if (payId == null) {
            DataMetricHolder.getInstance().increment(DataUtils.EVENT_A,
                MetricName.MISSING_PAY_ID);
            logger.logError("dropping records. invalid payId is coming.");
            return false;
          }
        return true;
    }

    private boolean isValidClientIdDeviceId() {
        if (clientId == null &amp;&amp; deviceId == null) {
            DataMetricHolder.getInstance().increment(DataUtils.EVENT_A,
                MetricName.INVALID_ID);
            logger.logError("dropping records. invalid clientId and deviceId is coming.");
            return false;
          }
        return true;
    }

    public String getClientId() {
        return clientId;
    }

    public String getDeviceId() {
        return deviceId;
    }

    public String getPayId() {
        return payId;
    }

    public String getCrossId() {
        return crossId;
    }

    public Map&lt;String, String&gt; getAppPayload() {
        return appPayload;
    }
}


Then your code might look something like: 

try {
  GenericRecordDomainDataDecoder decoder = new GenericRecordDomainDataDecoder(config);
  while (true) {
    ConsumerRecords&lt;byte[], byte[]&gt; records = consumer.poll(1000);
    for (ConsumerRecord&lt;byte[], byte[]&gt; record : records) {
      DataMetricHolder.getInstance().increment(DataUtils.EVENT_A,
          MetricName.TOTAL_NUMBER);
      GenericRecord payload = decoder.decode(record.value());

      //Replaced validation logic with the following 3 lines:
      PacketValidation validation = new PacketValidation(payload);
      if(!validation.isValid())
          continue;

      DataMetricHolder.getInstance().increment(validation.getPayId(), MetricName.TOTAL_NUMBER);

      Builder packetBuilder = new DataPacket.Builder(payload);
      if (validation.getClientId() != null) {
        packetBuilder.setClientId(validation.getClientId());
      }
      if (validation.getDeviceId() != null) {
        packetBuilder.setDeviceId(validation.getDeviceId());
      }
      if (validation.getCrossId() != null) {
        packetBuilder.setCrossId(validation.getCrossId());
      }
      DataPacket packet =
          packetBuilder.setPayId(validation.getPayId()).appPayload(validation.getAppPayload()).build();
      System.out.println(packet);
    }
  }
} catch (Exception ex) {
  // logging error
}


Some final thoughts: 

I'm not sure how often you use the field names, but it's generally a good idea to use constants of some kind if you're going to use them in more than one place. So for something like this: 

clientId = (String) DataUtils.parseRecord(payload, "clientId");


I would typically expect to see something more like: 

clientId = (String) DataUtils.parseRecord(payload, Constants.CLIENT_ID);


This helps to avoid spelling mistakes, whitespace issues, case sensitivity issues, etc.

I also noticed that your code has the following if statement: 

Map&lt;String, String&gt; appPayload...
if (MapUtils.isEmpty(appPayload))


I'm assuming that Map is java.util.Map in which case I'm not sure why you don't simply use Map.isEmpty() unless MapUtils is doing a null check at the same time - meaning treating null as being empty.

I hope this helps you! 

EDIT

Regarding your comment about having multiple consumers, I think you'll probably want to use a combination of the composite pattern and the strategy pattern so that you can try to reuse as much code as possible.

As for implementing a solution to this problem - if your consumers all use the GenericRecord class you could do something like the following: 

public class PacketValidation {
    //Added a collection of Validatable objects to the class
    private Collection&lt;Validatable&gt; validationList;
    private String clientId, deviceId, payId, crossId;
    private Map&lt;String,String&gt; appPayload;
    private Logger logger;
    private GenericRecord record;

    public PacketValidation(GenericRecord payload){
        record = payload;
        clientId = (String) DataUtils.parseRecord(payload, "clientId");
        deviceId = (String) DataUtils.parseRecord(payload, "deviceId");
        payId = (String) DataUtils.parseRecord(payload, "payId");
        crossId = (String) DataUtils.parseRecord(payload, "crossId");
        appPayload = (Map&lt;String, String&gt;) DataUtils.parseRecord(payload, "appPayload");

        //Added logic in constructor to create the validators
        // only created one as this is a quick example.
        validationList = new ArrayList&lt;Validatable&gt;();
        validationList.add(new ClientIdAndDeviceIdValidator());
    }

    //Determine if the data is valid by calling `isValid` on all validators
    public boolean isValid(){
        for(Validatable validate: validationList){
            if(!validate.isValid(record))
                return false;
        }
        return true;
    }

    //Getters removed for brevity
}


The validator class would look something like: 

public class ClientIdAndDeviceIdValidator implements Validatable {

    // Again, not sure how you're initializing this so just added a field for now.
    private Logger logger;

    @Override
    public boolean isValid(GenericRecord record) {
        String clientId = (String) DataUtils.parseRecord(record, "clientId");
        String deviceId = (String) DataUtils.parseRecord(record, "deviceId");

        if (clientId == null &amp;&amp; deviceId == null) {
            DataMetricHolder.getInstance().increment(DataUtils.EVENT_A,
                    MetricName.INVALID_ID);
            logger.logError("dropping records. invalid clientId and deviceId is coming.");
            return false;
        }
        return true;
    }

}


and the Validatable interface may look like this: 

public interface Validatable {
    public boolean isValid(GenericRecord record);
}


Now what you'll notice is that the implementation, ClientIdAndDeviceIdValidator, has to get the fields out of the GenericRecord because of the way the isValid method is written. Sometimes you want to validate that two fields together and other times just one field so in order to process all of the Validator objects generically we have to go one level above the fields and use the GenericRecord. Also note that I'm assuming you'll have one class for each consumer - therefore PacketValidation is not a generic class but rather one that is specific to the consumer.

Another possible solution would involve using Java Generics; here is an example: 

public class PacketValidationGeneric {
    //Added a generic validatable object field
    private GenericValidatable&lt;String&gt; clientIdDeviceIdValidator;
    private String clientId, deviceId, payId, crossId;
    private Map&lt;String,String&gt; appPayload;
    private Logger logger;

    public PacketValidationGeneric(GenericRecord payload){
        clientId = (String) DataUtils.parseRecord(payload, "clientId");
        deviceId = (String) DataUtils.parseRecord(payload, "deviceId");
        payId = (String) DataUtils.parseRecord(payload, "payId");
        crossId = (String) DataUtils.parseRecord(payload, "crossId");
        appPayload = (Map&lt;String, String&gt;) DataUtils.parseRecord(payload, "appPayload");

        //Creating the specific validator here:
        clientIdDeviceIdValidator = new ClientIdAndDeviceIdGenericValidator();
    }

    public boolean isValid(){
        //Obviously there is more to be done that just validate the two fields, but this is just a quick example.
        return clientIdDeviceIdValidator.isValid(clientId, deviceId);
    }

    //Removed some code for brevity
}


Here is the validator implementation:

public class ClientIdAndDeviceIdGenericValidator implements GenericValidatable&lt;String&gt; {

    // Again, not sure how you're initializing this so just added a field for now.
    private Logger logger;

    @Override
    public boolean isValid(String ... fieldList) {
        //Since the parameter is very generic, you need to validate the input.
        if(fieldList.length &lt; 2)
            throw new RuntimeException("Not enough fields provided, minimum 2 fields required - clientId and deviceId in that order.");
        String clientId = fieldList[0];
        String deviceId = fieldList[1];

        if (clientId == null &amp;&amp; deviceId == null) {
            DataMetricHolder.getInstance().increment(DataUtils.EVENT_A,
                    MetricName.INVALID_ID);
            logger.logError("dropping records. invalid clientId and deviceId is coming.");
            return false;
        }
        return true;
    }

}


This is the interface: 

public interface GenericValidatable&lt;T&gt; {
    public boolean isValid(T ... fieldList);
}


You'll notice that the PacketValidationGeneric.isValid() method can no longer invoke each validator in the same way because the parameters can differ. However, if you're going to have one class for each consumer then maybe it doesn't matter so much that you have to invoke each validator this way. Also notice, however, that you need to validate the input to each validator as the parameter to GenericValidatable.isValid is generic.

Once again, I hope this helps!