class NotificationBolt extends BaseTickTupleAwareRichBolt {

    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        if (!stormConf.containsKey(CATALOG_ROOT_URL)) {
            throw new IllegalArgumentException("conf must contain " + CATALOG_ROOT_URL);
        }

        Map<String, Object> notificationConf = null;
        if (stormConf.get(NOTIFICATION_SERVICE_CONFIG_KEY) != null) {
            notificationConf = (Map<String, Object>) stormConf.get(NOTIFICATION_SERVICE_CONFIG_KEY);
        } else {
            notificationConf = Collections.emptyMap();
        }

        NotificationStore notificationStore = null;
        try {
            if (!StringUtils.isEmpty(notificationStoreClazz)) {
                Class<?> clazz = Class.forName(notificationStoreClazz);
                notificationStore = (NotificationStore) clazz.newInstance();
                Map<String, Object> config = (Map<String, Object>) stormConf.get(NOTIFICATION_STORE_CONFIG_KEY);
                if (config == null) {
                    config = Collections.emptyMap();
                }
                notificationStore.init(config);
            }
        } catch (ClassNotFoundException | InstantiationException| IllegalAccessException ex) {
            throw new RuntimeException(ex);
        }

        notificationService = new NotificationServiceImpl(notificationConf, notificationStore);

        String jarPath = "";
        if (stormConf.containsKey(LOCAL_NOTIFIER_JAR_PATH)) {
            jarPath = String.format("%s%s%s", stormConf.get(LOCAL_NOTIFIER_JAR_PATH).toString(),
                    File.separator, notificationSink.getNotifierJarFileName());
        }

        Properties props = new Properties();
        props.putAll(convertMapValuesToString(notificationSink.getNotifierProperties()));
        NotifierConfig notifierConfig = new NotifierConfigImpl(props,
                convertMapValuesToString(notificationSink.getNotifierFieldValues()),
                notificationSink.getNotifierClassName(), jarPath);

        notificationContext = new BoltNotificationContext(collector, notifierConfig);
        notificationService.register(notificationSink.getNotifierName(), notificationContext);
    }

    public  NotificationBolt(String notificationSinkJson);
    public  NotificationBolt(NotificationSink notificationSink);

}