/*
 * Copyright 2009-2013 by The Regents of the University of California
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * you may obtain a copy of the License from
 * 
 *     http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package edu.uci.ics.pregelix.api.util;

import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.Counters;

import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.pregelix.api.graph.GlobalAggregator;
import edu.uci.ics.pregelix.api.graph.Vertex;
import edu.uci.ics.pregelix.api.io.WritableSizable;

/**
 * A global aggregator that produces a Hadoop mapreduce.Counters object
 *
 */
public abstract class HadoopCountersAggregator<I extends WritableComparable, V extends Writable, E extends Writable, M extends WritableSizable, P extends Writable> extends
        GlobalAggregator<I, V, E, M, P, Counters> {

}

@@ -669,16 +669,105 @@
 .util;%0A%0A
+import java.io.DataInputStream;%0Aimport java.io.IOException;%0Aimport java.io.InputStream;%0A%0A
 import o
@@ -792,24 +792,24 @@
 o.Writable;%0A
-
 import org.a
@@ -1189,15 +1189,42 @@
 ect%0A
- *%0A
  */%0A
+@SuppressWarnings(%22rawtypes%22)%0A
 publ
@@ -1382,24 +1382,16 @@
 ritable%3E
- extends
 %0A       
@@ -1391,16 +1391,24 @@
         
+extends 
 GlobalAg
@@ -1432,22 +1432,1401 @@
  M, 
-P, Counters%3E %7B%0A
+Counters, Counters%3E %7B%0A    private ResettableCounters counters = new ResettableCounters();%0A%0A    public Counters getCounters() %7B%0A        return counters;%0A    %7D%0A%0A    @Override%0A    public void init() %7B%0A        counters.reset();%0A    %7D%0A%0A    @Override%0A    public void step(Counters partialResult) %7B%0A        counters.incrAllCounters(partialResult);%0A    %7D%0A%0A    @Override%0A    public Counters finishPartial() %7B%0A        return counters;%0A    %7D%0A%0A    @Override%0A    public Counters finishFinal() %7B%0A        return counters;%0A    %7D%0A%0A    /**%0A     * mapreduce.Counters object that is resettable via .reset()%0A     */%0A    public static class ResettableCounters extends Counters %7B%0A        private static final DataInputStream zeroStream = new DataInputStream(new InputStream() %7B%0A            @Override%0A            public int read() throws IOException %7B%0A                return 0;%0A            %7D%0A        %7D);%0A%0A        /**%0A         * Reset this Counters object%0A         * %0A         * The reset is done by simulating a readFields() from a stream of 0's,%0A         * indicating a serialized length of 0 groups. The Counters' cache is not changed. %0A         */%0A        public void reset() %7B%0A            try %7B%0A                this.readFields(zeroStream);%0A            %7D catch (IOException e) %7B%0A                throw new RuntimeException(%22Unexpected failure when trying to reset Counters object!%22, e);%0A            %7D%0A        %7D%0A    %7D
 %0A%7D%0A

use Counters as partial value to simplify HadoopCountersAggregator