import org.elasticsearch.common.joda.time.format.DateTimeFormatter; //导入依赖的package包/类
@Override
public Event intercept(Event event) {
    Map<String, String> headers = event.getHeaders();

    if(headers.containsKey(timestampStr)) {
        String timestamp = headers.get(timestampStr);
        Long millis = Long.parseLong(timestamp);
        DateTimeFormatter fmt = ISODateTimeFormat.dateTime();
        headers.put(timestampStr, fmt.print(millis));

        // elasticsearch sink will look for a timestamp in milliseconds
        headers.put("timestamp", timestamp);
    }

    return event;
}
 

import org.elasticsearch.common.joda.time.format.DateTimeFormatter; //导入依赖的package包/类
/**
 * Method for accessing date formatter objects from cached store.
 * @param format of date with which the formatter will work
 * @return formatter object
 */
protected static DateTimeFormatter getDateTimeFormatterByFormat( String format ) {
	
	if (dateTimeFormattersCache.containsKey(format)) {
        return dateTimeFormattersCache.get(format);
    } else {
        DateTimeFormatter dtf = DateTimeFormat.forPattern(format).withLocale(Locale.US).withZone(DateTimeZone.UTC);
        dateTimeFormattersCache.put(format, dtf);
        return dtf;
    }
	
}