wherehows-kafka/src/main/java/wherehows/processors/MetadataChangeProcessor.java
public void process(IndexedRecord indexedRecord) throws Exception {

    if (indexedRecord == null || indexedRecord.getClass() != MetadataChangeEvent.class) {
      throw new IllegalArgumentException("Invalid record");
    }

    log.debug("Processing Metadata Change Event record. ");

    MetadataChangeEvent record = (MetadataChangeEvent) indexedRecord;

    final KafkaAuditHeader auditHeader = record.auditHeader;
    if (auditHeader == null) {
      log.warn("MetadataChangeEvent without auditHeader, abort process. " + record.toString());
      return;
    }

    final DatasetIdentifier identifier = record.datasetIdentifier;
    log.info("MCE: " + identifier); // TODO: remove. For debugging only
    final ChangeAuditStamp changeAuditStamp = record.changeAuditStamp;
    final ChangeType changeType = changeAuditStamp.type;

    if (changeType == ChangeType.DELETE) {
      // TODO: delete dataset
      log.debug("Dataset Deleted: " + identifier);
      return;
    }

    // create or update dataset
    DictDataset ds =
        _dictDatasetDao.insertUpdateDataset(identifier, changeAuditStamp, record.datasetProperty, record.schema,
            record.deploymentInfo, toStringList(record.tags), record.capacity, record.partitionSpec);

    // if schema is not null, insert or update schema
    if (record.schema != null) {
      _fieldDetailDao.insertUpdateDatasetFields(identifier, ds.getId(), record.datasetProperty, changeAuditStamp,
          record.schema);
    }

    // if owners are not null, insert or update owner
    if (record.owners != null) {
      _ownerDao.insertUpdateOwnership(identifier, ds.getId(), changeAuditStamp, record.owners);
    }

    // if compliance is not null, insert or update compliance
    if (record.compliancePolicy != null) {
      _complianceDao.insertUpdateCompliance(identifier, ds.getId(), changeAuditStamp, record.compliancePolicy);
    }

    // if suggested compliance is not null, insert or update suggested compliance
    if (record.suggestedCompliancePolicy != null) {
      _complianceDao.insertUpdateSuggestedCompliance(identifier, ds.getId(), changeAuditStamp,
          record.suggestedCompliancePolicy);
    }
  }
/**
   * Process a MetadataChangeEvent record
   * @param indexedRecord IndexedRecord
   * @throws Exception
   */
Process a MetadataChangeEvent record
@param indexedRecord IndexedRecord
@throws Exception
Process a MetadataChangeEvent record @param indexedRecord IndexedRecord @throws Exception
public void process(IndexedRecord indexedRecord) throws Exception {
    if (indexedRecord == null || indexedRecord.getClass() != MetadataChangeEvent.class) {
      throw new IllegalArgumentException("Invalid record");
    }
    log.debug("Processing Metadata Change Event record. ");
    MetadataChangeEvent record = (MetadataChangeEvent) indexedRecord;
    final KafkaAuditHeader auditHeader = record.auditHeader;
    if (auditHeader == null) {
      log.warn("MetadataChangeEvent without auditHeader, abort process. " + record.toString());
      return;
    }
    final DatasetIdentifier identifier = record.datasetIdentifier;
    log.info("MCE: " + identifier); 
    final ChangeAuditStamp changeAuditStamp = record.changeAuditStamp;
    final ChangeType changeType = changeAuditStamp.type;
    if (changeType == ChangeType.DELETE) {
      log.debug("Dataset Deleted: " + identifier);
      return;
    }
    DictDataset ds =
        _dictDatasetDao.insertUpdateDataset(identifier, changeAuditStamp, record.datasetProperty, record.schema,
            record.deploymentInfo, toStringList(record.tags), record.capacity, record.partitionSpec);
    if (record.schema != null) {
      _fieldDetailDao.insertUpdateDatasetFields(identifier, ds.getId(), record.datasetProperty, changeAuditStamp,
          record.schema);
    }
    if (record.owners != null) {
      _ownerDao.insertUpdateOwnership(identifier, ds.getId(), changeAuditStamp, record.owners);
    }
    if (record.compliancePolicy != null) {
      _complianceDao.insertUpdateCompliance(identifier, ds.getId(), changeAuditStamp, record.compliancePolicy);
    }
    if (record.suggestedCompliancePolicy != null) {
      _complianceDao.insertUpdateSuggestedCompliance(identifier, ds.getId(), changeAuditStamp,
          record.suggestedCompliancePolicy);
    }
  }
public void process ( IndexedRecord indexedRecord ) throws Exception { if ( indexedRecord == null || indexedRecord . getClass ( ) != MetadataChangeEvent . class ) { throw new IllegalArgumentException ( "Invalid record" ) ; } log . debug ( "Processing Metadata Change Event record. " ) ; MetadataChangeEvent record = ( MetadataChangeEvent ) indexedRecord ; final KafkaAuditHeader auditHeader = record . auditHeader ; if ( auditHeader == null ) { log . warn ( "MetadataChangeEvent without auditHeader, abort process. " + record . toString ( ) ) ; return ; } final DatasetIdentifier identifier = record . datasetIdentifier ; log . info ( "MCE: " + identifier ) ; final ChangeAuditStamp changeAuditStamp = record . changeAuditStamp ; final ChangeType changeType = changeAuditStamp . type ; if ( changeType == ChangeType . DELETE ) { log . debug ( "Dataset Deleted: " + identifier ) ; return ; } DictDataset ds = _dictDatasetDao . insertUpdateDataset ( identifier , changeAuditStamp , record . datasetProperty , record . schema , record . deploymentInfo , toStringList ( record . tags ) , record . capacity , record . partitionSpec ) ; if ( record . schema != null ) { _fieldDetailDao . insertUpdateDatasetFields ( identifier , ds . getId ( ) , record . datasetProperty , changeAuditStamp , record . schema ) ; } if ( record . owners != null ) { _ownerDao . insertUpdateOwnership ( identifier , ds . getId ( ) , changeAuditStamp , record . owners ) ; } if ( record . compliancePolicy != null ) { _complianceDao . insertUpdateCompliance ( identifier , ds . getId ( ) , changeAuditStamp , record . compliancePolicy ) ; } if ( record . suggestedCompliancePolicy != null ) { _complianceDao . insertUpdateSuggestedCompliance ( identifier , ds . getId ( ) , changeAuditStamp , record . suggestedCompliancePolicy ) ; } }
Process a MetadataChangeEvent record
 @param indexedRecord IndexedRecord
 @throws Exception
Process a MetadataChangeEvent record @param indexedRecord IndexedRecord @throws Exception
// TODO: remove. For debugging only // TODO: delete dataset // create or update dataset // if schema is not null, insert or update schema // if owners are not null, insert or update owner // if compliance is not null, insert or update compliance // if suggested compliance is not null, insert or update suggested compliance