@EnableBinding({Sink.class})
public class EmailSink {
   private static final Logger LOGGER = LoggerFactory.getLogger(EmailSink.class);

   @StreamListener(target = Sink.INPUT, condition = "headers['notificationPriority']=='normal'")
   public void sentEmail(Message<Map<String, Object>> paymentNotification) {
       String status = String.valueOf(paymentNotification.getPayload().getOrDefault("status", "pending"));
       LOGGER.info("Normal Payment =" + status);
   }

   @StreamListener(target = Sink.INPUT, condition = "headers['notificationPriority']=='urgent'")
   public void sentUrgentEmail(Message<Map<String, Object>> paymentNotification) {
       String status = String.valueOf(paymentNotification.getPayload().getOrDefault("status", "pending"));
       LOGGER.info("Urgent Payment =" + status);
   }
}