### Question:
LocalityGroupPartitioner extends Partitioner<BulkIngestKey,Value> implements DelegatePartitioner { @Override public int getPartition(BulkIngestKey bKey, Value value, int reducers) { Text columnFamily = bKey.getKey().getColumnFamily(); int index = colFams.indexOf(columnFamily); if (index == -1) { if (!hasSeenUnknownColFams) { log.warn("Unexpected column family: " + columnFamily); hasSeenUnknownColFams = true; } return colFams.size() % reducers; } return index % reducers; }  @Override int getPartition(BulkIngestKey bKey, Value value, int reducers); @Override Configuration getConf(); @Override void setConf(Configuration conf); @Override void configureWithPrefix(String prefix); @Override int getNumPartitions(); @Override void initializeJob(Job job); static final String COLUMN_FAMILIES; }### Answer:
@Test public void testGetPartition() throws Exception { Configuration conf = new Configuration(); String tableName = "tableX"; conf.set(tableName + "." + LocalityGroupPartitioner.COLUMN_FAMILIES, "CITY,STATE"); LocalityGroupPartitioner partitioner = LocalityGroupPartitioner.class.newInstance(); partitioner.setConf(conf); partitioner.configureWithPrefix(tableName); Text tbl = new Text("table"); int expectedReducerCity = 0; int expectedReducerState = 1; for (int i = 0; i < 10; ++i) { String shard = "" + i; for (int j = 0; j < 2; ++j) { Assert.assertEquals(" failed on row " + shard + " column family CITY", expectedReducerCity, partitioner.getPartition(new BulkIngestKey(tbl, new Key(shard, "CITY", Integer.toHexString(j))), null, NUM_REDUCERS)); Assert.assertEquals(" failed on row " + shard + " column family STATE", expectedReducerState, partitioner.getPartition(new BulkIngestKey(tbl, new Key(shard, "STATE", Integer.toHexString(j))), null, NUM_REDUCERS)); } } }