@Test
public void testColumnBasedIndexing() throws Exception {

    IndexerConf conf = new IndexerConfBuilder().table(TABLE_A).mappingType(IndexerConf.MappingType.COLUMN).build();
    ResultToSolrMapper mapper = new ResultToSolrMapper() {
        @Override
        public boolean isRelevantKV(KeyValue kv) {
            return Bytes.toString(CellUtil.cloneFamily(kv)).equals("messages");
        }
        @Override
        public Get getGet(byte[] row) {
            return new Get(row);
        }
        @Override
        public void map(Result result, SolrUpdateWriter solrUpdateWriter) {
            solrUpdateWriter.add(new SolrInputDocument());
        }
        @Override
        public boolean containsRequiredData(Result result) {
            throw new UnsupportedOperationException("Not implemented");
        }
    };
    Indexer indexer = Indexer.createIndexer("index name", conf, TABLE_A, mapper, null, null, solrDocumentWriter);
    IndexingEventListener indexingEventListener = new IndexingEventListener(indexer, TABLE_A, false);
    List<Cell> kvs = Lists.newArrayList(
        (Cell)new KeyValue(Bytes.toBytes("row1"), Bytes.toBytes("messages"), Bytes.toBytes("msg1"),
                    Bytes.toBytes("the message")), new KeyValue(Bytes.toBytes("row1"), Bytes.toBytes("messages"),
                    Bytes.toBytes("msg2"), Bytes.toBytes("another message")));
    SepEvent event = new SepEvent(Bytes.toBytes(TABLE_A), Bytes.toBytes("row1"), kvs, null);
    indexingEventListener.processEvents(Collections.singletonList(event));
    ArgumentCaptor<Map> arg = ArgumentCaptor.forClass(Map.class);
    verify(solrDocumentWriter).add(eq(-1), arg.capture());
    Map<String,SolrInputDocument> docs = arg.getValue();
    assertEquals(2, docs.size());
    Set<String> documentIds = Sets.newHashSet(Collections2.transform(docs.values(), new Function<SolrInputDocument,String>(){
        @Override
        public String apply(@Nullable SolrInputDocument input) {
            return (String)input.getFieldValue("id");
        }}));
    assertEquals(Sets.newHashSet("row1-messages-msg1","row1-messages-msg2"), documentIds);
}