return cachedData == null;
}
@Override
public Map<String, PartitionTimestamps> putData(Map<String, Map<Long, Map<String, String>>> value, TimeUnit timeUnit) {
final ElapsedTimer timer = new ElapsedTimer();
timer.startInterval();
Map<String, PartitionTimestamps> timestamps = new HashMap<String, PartitionTimestamps>();
Map<String, TreeMap<Long, Map<String, String>>> cachedData = getCachedData();
for (Entry<String, Map<Long, Map<String, String>>> entry : value.entrySet()) {
String feedID = entry.getKey();
long largestTime = 0;
long smallestTime = 0;
synchronized (this) {
TreeMap<Long, Map<String, String>> cachedFeedData = cachedData.get(feedID);
if (cachedFeedData == null) {
cachedFeedData = new TreeMap<Long, Map<String, String>>(TIMESTAMP_COMPARATOR);
cachedData.put(feedID, cachedFeedData);
}
for (Entry<Long, Map<String, String>> feedData : entry.getValue().entrySet()) {
Long time = feedData.getKey();
time = TimeUnit.NANOSECONDS.convert(time, timeUnit);
LOGGER.debug("Putting data for feed {} with time {}", feedID, time);
if (time.longValue() > largestTime) {
largestTime = time.longValue();
}
if (smallestTime == 0) {
smallestTime = time.longValue();
} else if (time.longValue() < smallestTime) {
smallestTime = time.longValue();
}
Map<String, String> clonedFeedData = new HashMap<String, String>(feedData.getValue());
cachedFeedData.put(time, clonedFeedData);
}
}
timestamps.put(feedID, new PartitionTimestamps(smallestTime, largestTime));
}
timer.stopInterval();
WRITE_PERF_LOGGER.debug("Time to write {} feeds: {} from partition " + this.env.getCurrentBufferPartition(), value.size(), timer.getIntervalInMillis());
return timestamps;
}