code: func (plugin *ExamplePlugin) asyncEventHandler() {
  	plugin.Log.Info("Started Kafka async event handler...")
  	asyncSuccessCounter := 0
  	if messageCountNum == 0 {
  		plugin.asyncSuccess = true
  	}
  
  	for {
  		select {
  		// Channel subscribed with watcher
  		case message := <-plugin.asyncSubscription:
  			plugin.Log.Infof("Received async Kafka Message, topic '%s', partition '%v', offset '%v', key: '%s', ",
  				message.GetTopic(), message.GetPartition(), message.GetOffset(), message.GetKey())
  			// Note: mark the offset if required
  			if message.GetPartition() != asyncMessagePartition {
  				plugin.Log.Errorf("Received async message with unexpected partition: %v", message.GetOffset())
  			}
  			if message.GetOffset() < messageOffset {
  				plugin.Log.Errorf("Received async message with unexpected offset: %v", message.GetOffset())
  			}
  			// Success callback channel
  		case message := <-plugin.asyncSuccessChannel:
  			plugin.Log.Infof("Async message successfully delivered, topic '%s', partition '%v', offset '%v', key: '%s', ",
  				message.GetTopic(), message.GetPartition(), message.GetOffset(), message.GetKey())
  			// Note: mark the offset if required
  			asyncSuccessCounter++
  			if asyncSuccessCounter == messageCountNum {
  				plugin.asyncSuccess = true
  			}
  			// Error callback channel
  		case err := <-plugin.asyncErrorChannel:
  			plugin.Log.Errorf("Failed to publish async message, %v", err)
  		}
  	}
  }
docs: // asyncEventHandler is a Kafka consumer asynchronously processing events from
  // a channel associated with a specific topic, partition and a starting offset.
  // If a producer sends a message matching this destination criteria, the consumer
  // will receive it.