global without sharing class XXX_SSM13_ReCloseCheckBatch implements Database.Batchable<SObject>, Database.Stateful {
    global final static Set<String> fields2Check = new Set<String>{'MRC__c','CurrencyISOCode','Commited_Bandwidth__c',
            'Recommended_Base_Unit_Price__c','Recommended_Burst_Price__c'};
    global final static string query =
            'SELECT Id,MRC__c,CurrencyISOCode,Commited_Bandwidth__c,Recommended_Base_Unit_Price__c,' +
                    'Recommended_Burst_Price__c, ' +
                    '(SELECT Id,MRC__c,CurrencyISOCode,Commited_Bandwidth__c,Recommended_Base_Unit_Price__c,' +
                    'Recommended_Burst_Price__c, TECH_Definition_JSON__c ' +
                    'FROM Sold_Service_Modifications__r ' +
                    'WHERE Stage__c = \''+ApplicationConstant.SSM_STAGE_LIVE+'\' AND Action__c != \'Add\' ' +
                    'ORDER BY CreatedDate DESC LIMIT 1) FROM Sold_Service__c ' +
                    'WHERE Stage__c = \'Live Service\' AND Status__c = \'Active\'';

    global Integer totalCorruptedNo = 0;
    global Integer totalCorruptedJSONNo = 0;
    global Set<Id> corruptedSSMs = new Set<Id>();
    global Set<Id> corruptedSSMJSON = new Set<Id>();
    Map<Id,List<String>> fieldDiff = new Map<Id, List<String>>();

    global Database.QueryLocator start(Database.BatchableContext context) {
        return  Database.getQueryLocator(query);
    }

    global void execute(Database.BatchableContext context, List<Sold_Service__c> scope) {
        for(Sold_Service__c ss:scope) {
            Sold_Service_Modification__c ssm;
            if(ss.Sold_Service_Modifications__r.size() > 0) {
                ssm = ss.Sold_Service_Modifications__r[0];
            }

            if(ssm != null && areFieldsDifferent(ss,ssm) && ssm.TECH_Definition_JSON__c != null) {
                totalCorruptedNo++;
                corruptedSSMs.add(ssm.Id);
            }

            if(ssm != null && ssm.TECH_Definition_JSON__c == null) {
                totalCorruptedJSONNo++;
                corruptedSSMJSON.add(ssm.Id);

                String changedFields = '';
                String modDetails = '';
                Schema.DescribeSObjectResult obj = Schema.getGlobalDescribe().get('Sold_Service__c').getDescribe();
                Map<String, Schema.SObjectField> schemaFieldMap = obj.fields.getMap();
                if(newSS.Item_Action__c != CPQ_PRD04_SoldServiceMappingHlpr.ACTION_ADD)
                    for(String fieldName : schemaFieldMap.keySet()){
                        if(newSS.get(fieldName) != null && !ApplicationConstant.SOLDSERVICE_FIELDS_EXCLUDE.contains(fieldName) && areFieldsDifferent(ss, ssm)) {
                            ssResult.put(obj.Fields.getMap().get(fieldName).getDescribe().getName(), ssm.get(fieldName));
                        }
                    }
                }

                ssm.TECH_Definition_JSON__c = JSON.Serialize(ssResult);
                return result;


            }

        }
    }

    private Boolean areFieldsDifferent(Sold_Service__c ss,Sold_Service_Modification__c ssm) {
        Boolean different = false;
        for(String fieldName : fields2Check) {
            if(ss.get(fieldName) != ssm.get(fieldName)) {
                different = true;
                if(fieldDiff.get(ssm.Id) == null)
                    fieldDiff.put(ssm.Id,new List<String>{fieldName});
                else
                    fieldDiff.get(ssm.Id).add(fieldName);
            }
        }

        return different;
    }

    global void finish(Database.BatchableContext BC) {
        AsyncApexJob a = [Select Id, Status,ExtendedStatus,NumberOfErrors,     JobItemsProcessed,
                TotalJobItems, CreatedBy.Email
        from AsyncApexJob where Id =:BC.getJobId()];

        Messaging.SingleEmailMessage mail = new Messaging.SingleEmailMessage();
        String[] toAddresses = new String[] {a.CreatedBy.Email};
        mail.setToAddresses(toAddresses);
        mail.setSubject('Match Merge Batch ' + a.Status);
        String body = 'batches processed: ' + a.TotalJobItems +
                ' with: '+ a.NumberOfErrors + ' failures. '+totalCorruptedNo +
                ' corrupted SSMs found. Corrupted SSMs: \n';
        for(Id ssmId : fieldDiff.keyset()){
            body+= ssmId + ' \n Field Difference: \n';
            for(String ssmField : fieldDiff.get(ssmId)){
                body+= '\t ' + ssmField;
            }
            body+= '\n';

        }
        body+= '. \n ' + totalCorruptedJSONNo +
                ' corrupted SSMs  with no JSON found. Corrupted SSMs: \n'+CSP_REST_Common.idSet2Str(corruptedSSMJSON);
        mail.setPlainTextBody(body);
        Messaging.sendEmail(new Messaging.SingleEmailMessage[] { mail });
    }

}