code: func (plugin *ExamplePlugin) syncEventHandler() {
  	plugin.Log.Info("Started Kafka sync event handler...")
  
  	time.Sleep(1 * time.Second)
  
  	// Initialize sync watcher.
  	plugin.kafkaWatcher = plugin.Kafka.NewPartitionWatcher("example-part-watcher")
  
  	// The watcher is consuming messages on a custom partition and an offset.
  	// If there is a producer who stores message to the same partition under
  	// the same or a newer offset, the message will be consumed.
  	err := plugin.kafkaWatcher.WatchPartition(messaging.ToProtoMsgChan(plugin.subscription), topic1,
  		syncMessagePartition, syncMessageOffset)
  	if err != nil {
  		plugin.Log.Error(err)
  	}
  
  	// Producer sends several messages (set in syncMessageCount).
  	// Consumer should receive only messages from desired partition and offset.
  	messageCounter := 0
  	for message := range plugin.subscription {
  		plugin.Log.Infof("Received sync Kafka Message, topic '%s', partition '%v', offset '%v', key: '%s', ",
  			message.GetTopic(), message.GetPartition(), message.GetOffset(), message.GetKey())
  		messageCounter++
  		if message.GetPartition() != syncMessagePartition {
  			plugin.Log.Errorf("Received sync message with unexpected partition: %v", message.GetOffset())
  		}
  		if message.GetOffset() < syncMessageOffset {
  			plugin.Log.Errorf("Received sync message with unexpected offset: %v", message.GetOffset())
  		}
  		// For example purpose: let it know that this part of the example is done
  		if messageCounter >= (syncMessageCount - syncMessageOffset) {
  			plugin.syncCaseDone = true
  		}
  	}
  }
docs: /*************
   * Consumers *
   *************/
  // syncEventHandler is a Kafka consumer synchronously 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.