@Override
    public void setup(Context context) {
      LOG.info("Setting up " + getClass() + " mapper.");
      Configuration conf = context.getConfiguration();
      cfRenameMap = createCfRenameMap(conf);
      filter = instantiateFilter(conf);
      String durabilityStr = conf.get(WAL_DURABILITY);
      if (durabilityStr != null) {
        durability = Durability.valueOf(durabilityStr.toUpperCase(Locale.ROOT));
        LOG.info("setting WAL durability to " + durability);
      } else {
        LOG.info("setting WAL durability to default.");
      }
      // TODO: This is kind of ugly doing setup of ZKW just to read the clusterid.
      ZKWatcher zkw = null;
      Exception ex = null;
      try {
        zkw = new ZKWatcher(conf, context.getTaskAttemptID().toString(), null);
        clusterIds = Collections.singletonList(ZKClusterId.getUUIDForCluster(zkw));
      } catch (ZooKeeperConnectionException e) {
        ex = e;
        LOG.error("Problem connecting to ZooKeper during task setup", e);
      } catch (KeeperException e) {
        ex = e;
        LOG.error("Problem reading ZooKeeper data during task setup", e);
      } catch (IOException e) {
        ex = e;
        LOG.error("Problem setting up task", e);
      } finally {
        if (zkw != null) zkw.close();
      }
      if (clusterIds == null) {
        // exit early if setup fails
        throw new RuntimeException(ex);
      }
    }
@param row     The current table row key.
@param value   The columns.
@param context The current context.
@throws IOException When something is broken with the data.
hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/Import.java