class InfinispanRemoteClient extends DB {

  private static final Log LOGGER = LogFactory.getLog(InfinispanRemoteClient.class);

  private RemoteCacheManager remoteIspnManager;
  private String cacheName = null;

  @Override
  public void init() throws DBException {
    remoteIspnManager = RemoteCacheManagerHolder.getInstance(getProperties());
    cacheName = getProperties().getProperty("cache");
  }

  @Override
  public void cleanup() {
    remoteIspnManager.stop();
    remoteIspnManager = null;
  }

  @Override
  public Status insert(String table, String recordKey, Map<String, ByteIterator> values) {
    String compositKey = createKey(table, recordKey);
    Map<String, String> stringValues = new HashMap<>();
    StringByteIterator.putAllAsStrings(stringValues, values);
    try {
      cache().put(compositKey, stringValues);
      return Status.OK;
    } catch (Exception e) {
      LOGGER.error(e);
      return Status.ERROR;
    }
  }

  @Override
  public Status read(String table, String recordKey, Set<String> fields, Map<String, ByteIterator> result) {<FILL_FUNCTION_BODY>}

  @Override
  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.NOT_IMPLEMENTED;
  }

  @Override
  public Status update(String table, String recordKey, Map<String, ByteIterator> values) {
    String compositKey = createKey(table, recordKey);
    try {
      Map<String, String> stringValues = new HashMap<>();
      StringByteIterator.putAllAsStrings(stringValues, values);
      cache().put(compositKey, stringValues);
      return Status.OK;
    } catch (Exception e) {
      LOGGER.error(e);
      return Status.ERROR;
    }
  }

  @Override
  public Status delete(String table, String recordKey) {
    String compositKey = createKey(table, recordKey);
    try {
      cache().remove(compositKey);
      return Status.OK;
    } catch (Exception e) {
      LOGGER.error(e);
      return Status.ERROR;
    }
  }

  private RemoteCache<String, Map<String, String>> cache() {
    if (this.cacheName != null) {
      return remoteIspnManager.getCache(cacheName);
    } else {
      return remoteIspnManager.getCache();
    }
  }

  private String createKey(String table, String recordKey) {
    return table + "-" + recordKey;
  }
}

    String compositKey = createKey(table, recordKey);
    try {
      Map<String, String> values = cache().get(compositKey);

      if (values == null || values.isEmpty()) {
        return Status.NOT_FOUND;
      }

      if (fields == null) { //get all field/value pairs
        StringByteIterator.putAllAsByteIterators(result, values);
      } else {
        for (String field : fields) {
          String value = values.get(field);
          if (value != null) {
            result.put(field, new StringByteIterator(value));
          }
        }
      }

      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