/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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 at
 *
 * 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 backtype.storm.grouping;

import java.io.Serializable;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import backtype.storm.generated.GlobalStreamId;
import backtype.storm.grouping.CustomStreamGrouping;
import backtype.storm.task.WorkerTopologyContext;
import backtype.storm.tuple.Fields;

import com.google.common.hash.HashFunction;
import com.google.common.hash.Hashing;

public class PartialKeyGrouping implements CustomStreamGrouping, Serializable {
    private static final long serialVersionUID = -447379837314000353L;
    private List<Integer> targetTasks;
    private long[] targetTaskStats;
    private HashFunction h1 = Hashing.murmur3_128(13);
    private HashFunction h2 = Hashing.murmur3_128(17);
    private Fields fields = null;
    private Fields outFields = null;

    public PartialKeyGrouping() {
        //Empty
    }

    public PartialKeyGrouping(Fields fields) {
        this.fields = fields;
    }

    @Override
    public void prepare(WorkerTopologyContext context, GlobalStreamId stream, List<Integer> targetTasks) {
        this.targetTasks = targetTasks;
        targetTaskStats = new long[this.targetTasks.size()];
        if (this.fields != null) {
            this.outFields = context.getComponentOutputFields(stream);
        }
    }

    @Override
    public List<Integer> chooseTasks(int taskId, List<Object> values) {
        List<Integer> boltIds = new ArrayList<Integer>(1);
        if (values.size() > 0) {
            byte[] raw = null;
            if (fields != null) {
                List<Object> selectedFields = outFields.select(fields, values);
                ByteBuffer out = ByteBuffer.allocate(selectedFields.size() * 4);
                for (Object o: selectedFields) {
                    if (o instanceof Object[]) {
                        out.putInt(Arrays.deepHashCode((Object[])o));
                    } else {
                        out.putInt(o.hashCode());
                    }
                }
                raw = out.array();
            } else {
                raw = values.get(0).toString().getBytes(); // assume key is the first field
            }
            int firstChoice = (int) (Math.abs(h1.hashBytes(raw).asLong()) % this.targetTasks.size());
            int secondChoice = (int) (Math.abs(h2.hashBytes(raw).asLong()) % this.targetTasks.size());
            int selected = targetTaskStats[firstChoice] > targetTaskStats[secondChoice] ? secondChoice : firstChoice;
            boltIds.add(targetTasks.get(selected));
            targetTaskStats[selected]++;
        }
        return boltIds;
    }
}

@@ -2620,16 +2620,146 @@
 tanceof 
+List) %7B%0A                        out.putInt(Arrays.deepHashCode(((List)o).toArray()));%0A                    %7D else if (o instanceof 
 Object%5B%5D
@@ -2850,32 +2850,57 @@
           %7D else
+ if (o instanceof byte%5B%5D)
  %7B%0A             
@@ -2925,20 +2925,1040 @@
 Int(
-o.hashCode()
+Arrays.hashCode((byte%5B%5D) o));%0A                    %7D else if (o instanceof short%5B%5D) %7B%0A                        out.putInt(Arrays.hashCode((short%5B%5D) o));%0A                    %7D else if (o instanceof int%5B%5D) %7B%0A                        out.putInt(Arrays.hashCode((int%5B%5D) o));%0A                    %7D else if (o instanceof long%5B%5D) %7B%0A                        out.putInt(Arrays.hashCode((long%5B%5D) o));%0A                    %7D else if (o instanceof char%5B%5D) %7B%0A                        out.putInt(Arrays.hashCode((char%5B%5D) o));%0A                    %7D else if (o instanceof float%5B%5D) %7B%0A                        out.putInt(Arrays.hashCode((float%5B%5D) o));%0A                    %7D else if (o instanceof double%5B%5D) %7B%0A                        out.putInt(Arrays.hashCode((double%5B%5D) o));%0A                    %7D else if (o instanceof boolean%5B%5D) %7B%0A                        out.putInt(Arrays.hashCode((boolean%5B%5D) o));%0A                    %7D else if (o != null) %7B%0A                        out.putInt(o.hashCode());%0A                    %7D else %7B%0A                      out.putInt(0
 );%0A 

handle Lists and primitive array types