Fire Platform Events from Batch Apex using Database.RaisesPlatformEvents Interface

SK_AccountProcessBatch.cls

global class SK_AccountProcessBatch implements Database.Batchable<sObject>,Database.Stateful,Database.RaisesPlatformEvents{
    global List<string> errorRecordsId=new List<String>();
    global Database.QueryLocator start(Database.BatchableContext BC){
        return Database.getQueryLocator('Select id,name,Type,rating,industry from Account  Limit 1000');
    }
    
    global void execute(Database.BatchableContext BC, List<sObject> scope){
        List<Account> accountListForUpdate = new List<Account>();
        for(sobject s : scope){
            Account acc =(Account)s;
            if(acc.Type!=null){
            	if(acc.Type.equalsIgnoreCase('Customer - Direct'))
                s.put('Rating','Hot');
            	accountListForUpdate.add(acc);   
            }
            
        }
        
        if(accountListForUpdate.size()>0){
            Database.SaveResult[] srList = database.update(accountListForUpdate,false);
            // Inspect publishing result 
            integer recordIndex=0;
            // Iterate through each returned result
            for (Database.SaveResult  sr : srList) {
                if (!sr.isSuccess()) {
                    for(Database.Error err : sr.getErrors()) {
                        System.debug('The following error has occurred.'+err.getStatusCode() + ': ' + err.getMessage());
                        errorRecordsId.add(accountListForUpdate[recordIndex].id);
                    }
                }
                recordIndex++;
            }
        }
    }
    
    global void finish(Database.BatchableContext BC){
        system.debug('*****failed recordIds:'+string.join(errorRecordsId,','));
        // Create an instance of the Demo event 
        Demo_Event__e demoEvent = new Demo_Event__e(
            Event_Info__c='Account records failed during SK_AccountProcessBatch processing', 
            Is_Event_Valid__c=true, 
            Event_Publisher__c='SK_AccountProcessBatch',
            Records_Ids__c=string.join(errorRecordsId,','));
        // Call method to publish events
        Database.SaveResult sr = EventBus.publish(demoEvent);
        // Inspect publishing result 
        if (sr.isSuccess()) {
            System.debug('Successfully published event.');
        } else {
            for(Database.Error err : sr.getErrors()) {
                System.debug('Error returned: ' +  err.getStatusCode() +' - ' + err.getMessage());
            }
        }
    }
}

SK_DemoEventTrigger.apxt

trigger DemoEventTrigger  on Demo_Event__e (after insert) {
    for (Demo_Event__e event: Trigger.New) {
      if (event.Is_Event_Valid__c == true) {
           //perform logic based on event information
           system.debug('**Records_Ids__c passed through platform events:'+event.Records_Ids__c);
      }
    }
}