// Path: src/main/java/com/dataartisans/flink/cascading/runtime/stats/EnumStringConverter.java
// public class EnumStringConverter {
// 
// 	public static String enumToGroup(Enum e) {
// 		return e.getDeclaringClass().getName();
// 	}
// 
// 	/**
// 	 * Convert an Enum to the counter portion of its name.
// 	 *
// 	 * @param e
// 	 * @return
// 	 */
// 	public static String enumToCounter(Enum e) {
// 		return e.name();
// 	}
// 
// 	public static String mergeGroupCounter(String group, String counter) {
// 		return group + "->" + counter;
// 	}
// 
// 	public static boolean accInGroup(String group, String accKey) {
// 		return accKey.startsWith(group + "->");
// 	}
// 
// 	public static boolean accMatchesGroupCounter(String accKey, String group, String counter) {
// 		return accKey.equals(group + "->" + counter);
// 	}
// 
// 	public static String groupCounterToGroup(String groupKey) {
// 		return groupKey.split("->")[0];
// 	}
// 
// 	public static String groupCounterToCounter(String groupKey) {
// 		return groupKey.split("->")[1];
// 	}
// 
// }
// 
// Path: src/main/java/com/dataartisans/flink/cascading/runtime/stats/AccumulatorCache.java
// public class AccumulatorCache {
// 
// 	private static final Logger LOG = LoggerFactory.getLogger(AccumulatorCache.class);
// 
// 	private JobID jobID;
// 
// 	private Client client;
// 
// 	private volatile Map<String, Object> currentAccumulators = Collections.emptyMap();
// 
// 	private final long updateIntervalMillis;
// 	private long lastUpdateTime;
// 
// 	public AccumulatorCache(int updateIntervalSecs) {
// 		this.updateIntervalMillis = updateIntervalSecs * 1000;
// 	}
// 
// 	public void update() {
// 		update(false);
// 	}
// 
// 	public void update(boolean force) {
// 
// 		long currentTime = System.currentTimeMillis();
// 		if (!force && currentTime - lastUpdateTime <= updateIntervalMillis) {
// 			return;
// 		}
// 
// 		if (jobID == null) {
// 			return;
// 		}
// 
// 		if (client != null) {
// 
// 			try {
// 				currentAccumulators = client.getAccumulators(jobID);
// 				lastUpdateTime = currentTime;
// 
// 				LOG.debug("Updated accumulators: {}", currentAccumulators);
// 			} catch (Exception e) {
// 				LOG.error("Failed to fetch accumulators for job {}.", jobID);
// 			}
// 
// 		}
// 
// 	}
// 
// 	public Map<String, Object> getCurrentAccumulators() {
// 		return currentAccumulators;
// 	}
// 
// 	public void setJobID(JobID jobID) {
// 		this.jobID = jobID;
// 	}
// 
// 	public void setClient(Client client) {
// 		this.client = client;
// 	}
// 
// 	public long getLastUpdateTime() {
// 		return lastUpdateTime;
// 	}
// }


import cascading.flow.FlowStep;
import cascading.management.state.ClientState;
import cascading.stats.FlowStepStats;
import com.dataartisans.flink.cascading.runtime.stats.EnumStringConverter;
import com.dataartisans.flink.cascading.runtime.stats.AccumulatorCache;
import java.util.Collection;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
		for (String key : currentAccumulators.keySet()) {
			result.add(EnumStringConverter.groupCounterToGroup(key));
		}
		return result;
	}

	@Override
	public Collection<String> getCountersFor(String group) {
		accumulatorCache.update();
		Map<String, Object> currentAccumulators = accumulatorCache.getCurrentAccumulators();
		Set<String> result = new HashSet<String>();

		for (String key : currentAccumulators.keySet()) {
			if (EnumStringConverter.accInGroup(group, key)) {
				result.add(EnumStringConverter.groupCounterToCounter(key));
			}
		}
		return result;
	}

	@Override
	public long getCounterValue(Enum counter) {
		return getCounterValue(EnumStringConverter.enumToGroup(counter), EnumStringConverter.enumToCounter(counter));
	}

	@Override
	public long getCounterValue(String group, String counter) {
		accumulatorCache.update();
		Map<String, Object> currentAccumulators = accumulatorCache.getCurrentAccumulators();

// Path: src/main/java/com/dataartisans/flink/cascading/runtime/stats/EnumStringConverter.java
// public class EnumStringConverter {
// 
// 	public static String enumToGroup(Enum e) {
// 		return e.getDeclaringClass().getName();
// 	}
// 
// 	/**
// 	 * Convert an Enum to the counter portion of its name.
// 	 *
// 	 * @param e
// 	 * @return
// 	 */
// 	public static String enumToCounter(Enum e) {
// 		return e.name();
// 	}
// 
// 	public static String mergeGroupCounter(String group, String counter) {
// 		return group + "->" + counter;
// 	}
// 
// 	public static boolean accInGroup(String group, String accKey) {
// 		return accKey.startsWith(group + "->");
// 	}
// 
// 	public static boolean accMatchesGroupCounter(String accKey, String group, String counter) {
// 		return accKey.equals(group + "->" + counter);
// 	}
// 
// 	public static String groupCounterToGroup(String groupKey) {
// 		return groupKey.split("->")[0];
// 	}
// 
// 	public static String groupCounterToCounter(String groupKey) {
// 		return groupKey.split("->")[1];
// 	}
// 
// }
// 
// Path: src/main/java/com/dataartisans/flink/cascading/runtime/stats/AccumulatorCache.java
// public class AccumulatorCache {
// 
// 	private static final Logger LOG = LoggerFactory.getLogger(AccumulatorCache.class);
// 
// 	private JobID jobID;
// 
// 	private Client client;
// 
// 	private volatile Map<String, Object> currentAccumulators = Collections.emptyMap();
// 
// 	private final long updateIntervalMillis;
// 	private long lastUpdateTime;
// 
// 	public AccumulatorCache(int updateIntervalSecs) {
// 		this.updateIntervalMillis = updateIntervalSecs * 1000;
// 	}
// 
// 	public void update() {
// 		update(false);
// 	}
// 
// 	public void update(boolean force) {
// 
// 		long currentTime = System.currentTimeMillis();
// 		if (!force && currentTime - lastUpdateTime <= updateIntervalMillis) {
// 			return;
// 		}
// 
// 		if (jobID == null) {
// 			return;
// 		}
// 
// 		if (client != null) {
// 
// 			try {
// 				currentAccumulators = client.getAccumulators(jobID);
// 				lastUpdateTime = currentTime;
// 
// 				LOG.debug("Updated accumulators: {}", currentAccumulators);
// 			} catch (Exception e) {
// 				LOG.error("Failed to fetch accumulators for job {}.", jobID);
// 			}
// 
// 		}
// 
// 	}
// 
// 	public Map<String, Object> getCurrentAccumulators() {
// 		return currentAccumulators;
// 	}
// 
// 	public void setJobID(JobID jobID) {
// 		this.jobID = jobID;
// 	}
// 
// 	public void setClient(Client client) {
// 		this.client = client;
// 	}
// 
// 	public long getLastUpdateTime() {
// 		return lastUpdateTime;
// 	}
// }


// Path: src/main/java/com/dataartisans/flink/cascading/planner/FlinkFlowStepStats.java
import cascading.flow.FlowStep;
import cascading.management.state.ClientState;
import cascading.stats.FlowStepStats;
import com.dataartisans.flink.cascading.runtime.stats.EnumStringConverter;
import com.dataartisans.flink.cascading.runtime.stats.AccumulatorCache;
import java.util.Collection;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

		for (String key : currentAccumulators.keySet()) {
			result.add(EnumStringConverter.groupCounterToGroup(key));
		}
		return result;
	}

	@Override
	public Collection<String> getCountersFor(String group) {
		accumulatorCache.update();
		Map<String, Object> currentAccumulators = accumulatorCache.getCurrentAccumulators();
		Set<String> result = new HashSet<String>();

		for (String key : currentAccumulators.keySet()) {
			if (EnumStringConverter.accInGroup(group, key)) {
				result.add(EnumStringConverter.groupCounterToCounter(key));
			}
		}
		return result;
	}

	@Override
	public long getCounterValue(Enum counter) {
		return getCounterValue(EnumStringConverter.enumToGroup(counter), EnumStringConverter.enumToCounter(counter));
	}

	@Override
	public long getCounterValue(String group, String counter) {
		accumulatorCache.update();
		Map<String, Object> currentAccumulators = accumulatorCache.getCurrentAccumulators();

		for (String key : currentAccumulators.keySet()) {