class InfinispanClient extends DB {
  private static final Log LOGGER = LogFactory.getLog(InfinispanClient.class);

  // An optimisation for clustered mode
  private final boolean clustered;

  private EmbeddedCacheManager infinispanManager;

  public InfinispanClient() {
    clustered = Boolean.getBoolean("infinispan.clustered");
  }

  public void init() throws DBException {
    try {
      infinispanManager = new DefaultCacheManager("infinispan-config.xml");
    } catch (IOException e) {
      throw new DBException(e);
    }
  }

  public void cleanup() {
    infinispanManager.stop();
    infinispanManager = null;
  }

  public Status read(String table, String key, Set<String> fields, Map<String, ByteIterator> result) {
    try {
      Map<String, String> row;
      if (clustered) {
        row = AtomicMapLookup.getAtomicMap(infinispanManager.getCache(table), key, false);
      } else {
        Cache<String, Map<String, String>> cache = infinispanManager.getCache(table);
        row = cache.get(key);
      }
      if (row != null) {
        result.clear();
        if (fields == null || fields.isEmpty()) {
          StringByteIterator.putAllAsByteIterators(result, row);
        } else {
          for (String field : fields) {
            result.put(field, new StringByteIterator(row.get(field)));
          }
        }
      }
      return Status.OK;
    } catch (Exception e) {
      LOGGER.error(e);
      return Status.ERROR;
    }
  }

  public Status scan(String table, String startkey, int recordcount,
      Set<String> fields, Vector<HashMap<String, ByteIterator>> result) {
    LOGGER.warn("Infinispan does not support scan semantics");
    return Status.OK;
  }

  public Status update(String table, String key, Map<String, ByteIterator> values) {
    try {
      if (clustered) {
        AtomicMap<String, String> row = AtomicMapLookup.getAtomicMap(infinispanManager.getCache(table), key);
        StringByteIterator.putAllAsStrings(row, values);
      } else {
        Cache<String, Map<String, String>> cache = infinispanManager.getCache(table);
        Map<String, String> row = cache.get(key);
        if (row == null) {
          row = StringByteIterator.getStringMap(values);
          cache.put(key, row);
        } else {
          StringByteIterator.putAllAsStrings(row, values);
        }
      }

      return Status.OK;
    } catch (Exception e) {
      LOGGER.error(e);
      return Status.ERROR;
    }
  }

  public Status insert(String table, String key, Map<String, ByteIterator> values) {<FILL_FUNCTION_BODY>}

  public Status delete(String table, String key) {
    try {
      if (clustered) {
        AtomicMapLookup.removeAtomicMap(infinispanManager.getCache(table), key);
      } else {
        infinispanManager.getCache(table).remove(key);
      }
      return Status.OK;
    } catch (Exception e) {
      LOGGER.error(e);
      return Status.ERROR;
    }
  }
}

    try {
      if (clustered) {
        AtomicMap<String, String> row = AtomicMapLookup.getAtomicMap(infinispanManager.getCache(table), key);
        row.clear();
        StringByteIterator.putAllAsStrings(row, values);
      } else {
        infinispanManager.getCache(table).put(key, values);
      }

      return Status.OK;
    } catch (Exception e) {
      LOGGER.error(e);
      return Status.ERROR;
    }
  
<methods>public non-sealed void <init>() ,public void cleanup() throws site.ycsb.DBException,public abstract site.ycsb.Status delete(java.lang.String, java.lang.String) ,public java.util.Properties getProperties() ,public void init() throws site.ycsb.DBException,public abstract site.ycsb.Status insert(java.lang.String, java.lang.String, Map<java.lang.String,site.ycsb.ByteIterator>) ,public abstract site.ycsb.Status read(java.lang.String, java.lang.String, Set<java.lang.String>, Map<java.lang.String,site.ycsb.ByteIterator>) ,public abstract site.ycsb.Status scan(java.lang.String, java.lang.String, int, Set<java.lang.String>, Vector<HashMap<java.lang.String,site.ycsb.ByteIterator>>) ,public void setProperties(java.util.Properties) ,public abstract site.ycsb.Status update(java.lang.String, java.lang.String, Map<java.lang.String,site.ycsb.ByteIterator>) <variables>private java.util.Properties properties