class DateIndexDataTypeHandler implements DataTypeHandler<KEYIN>, RawRecordMetadata {

    @Override
    public void setup(TaskAttemptContext context) {
        this.taskAttemptContext = context;
        this.conf = context.getConfiguration();
        this.shardIdFactory = new ShardIdFactory(conf);
        
        String tableName = conf.get(DATEINDEX_TNAME, null);
        if (null == tableName) {
            log.error(DATEINDEX_TNAME + " not specified, no date index will be created");
        } else {
            setDateIndexTableName(new Text(tableName));
        }
        
        this.dateIndexNumShards = conf.getInt(DATEINDEX_NUM_SHARDS, this.dateIndexNumShards);
        
        TypeRegistry registry = TypeRegistry.getInstance(conf);
        
        // instantiate the mapping of types to fields
        // now get the dates to be index for this datatype
        dataTypeToTypeToFields = new HashMap<>();
        for (Type dataType : registry.getTypes()) {
            Set<String> typeToFieldsSet = new HashSet<>();
            typeToFieldsSet.addAll(conf.getTrimmedStringCollection("all" + DATEINDEX_TYPE_TO_FIELDS));
            typeToFieldsSet.addAll(conf.getTrimmedStringCollection(dataType.typeName() + DATEINDEX_TYPE_TO_FIELDS));
            Multimap<String,String> typeToFields = HashMultimap.create();
            for (String typeToField : typeToFieldsSet) {
                String[] parts = StringUtils.split(typeToField, '=');
                if (parts.length != 2) {
                    throw new IllegalStateException("Improper date index type to field configuration: " + typeToField);
                }
                typeToFields.put(parts[0], parts[1]);
            }
            log.info(this.getClass().getSimpleName() + " configured for " + dataType.typeName() + ": " + typeToFields);
            dataTypeToTypeToFields.put(dataType.typeName(), typeToFields);
        }
    }

}