Not your code, but I notice that this uses Hadoop's ReflectionUtils#newInstance(Class<?>, Configuration).  I have some issues with this:

1. This method is meant to instantiate a class and then call it's setConf method with the provided configuration. None of the KeyValueStore instances implement Configurable and so will never have a setConf.  This doesn't cause an error, but you incur runtime checks each time this method is called as a result.
2. We pass in a new, bunk configuration that never gets used.  Since new Configuration() and not new Configuration(false) is used, we go through the overhead of creating a new Configuration with the default resources added for no reason.

Can we use vanilla reflection instead here?