class IgniteClient extends IgniteAbstractClient {
  /** */
  private static Logger log = LogManager.getLogger(IgniteClient.class);



  /** Cached binary type. */
  private BinaryType binType = null;
  /** Cached binary type's fields. */
  private final ConcurrentHashMap<String, BinaryField> fieldsCache = new ConcurrentHashMap<>();




  /**
   * Read a record from the database. Each field/value pair from the result will
   * be stored in a HashMap.
   *
   * @param table  The name of the table
   * @param key    The record key of the record to read.
   * @param fields The list of fields to read, or null for all of them
   * @param result A HashMap of field/value pairs for the result
   * @return Zero on success, a non-zero error code on error
   */
  @Override
  public Status read(String table, String key, Set<String> fields,
                     Map<String, ByteIterator> result) {
    try {
      BinaryObject po = cache.get(key);

      if (po == null) {
        return Status.NOT_FOUND;
      }

      if (binType == null) {
        binType = po.type();
      }

      for (String s : F.isEmpty(fields) ? binType.fieldNames() : fields) {
        BinaryField bfld = fieldsCache.get(s);

        if (bfld == null) {
          bfld = binType.field(s);
          fieldsCache.put(s, bfld);
        }

        String val = bfld.value(po);
        if (val != null) {
          result.put(s, new StringByteIterator(val));
        }

        if (debug) {
          log.info("table:{" + table + "}, key:{" + key + "}" + ", fields:{" + fields + "}");
          log.info("fields in po{" + binType.fieldNames() + "}");
          log.info("result {" + result + "}");
        }
      }

      return Status.OK;

    } catch (Exception e) {
      log.error(String.format("Error reading key: %s", key), e);

      return Status.ERROR;
    }
  }

  /**
   * Update a record in the database. Any field/value pairs in the specified
   * values HashMap will be written into the record with the specified record
   * key, overwriting any existing values with the same field name.
   *
   * @param table  The name of the table
   * @param key    The record key of the record to write.
   * @param values A HashMap of field/value pairs to update in the record
   * @return Zero on success, a non-zero error code on error
   */
  @Override
  public Status update(String table, String key,
                       Map<String, ByteIterator> values) {
    try {
      cache.invoke(key, new Updater(values));

      return Status.OK;
    } catch (Exception e) {
      log.error(String.format("Error updating key: %s", key), e);

      return Status.ERROR;
    }
  }

  /**
   * Insert a record in the database. Any field/value pairs in the specified
   * values HashMap will be written into the record with the specified record
   * key.
   *
   * @param table  The name of the table
   * @param key    The record key of the record to insert.
   * @param values A HashMap of field/value pairs to insert in the record
   * @return Zero on success, a non-zero error code on error
   */
  @Override
  public Status insert(String table, String key,
                       Map<String, ByteIterator> values) {<FILL_FUNCTION_BODY>}

  /**
   * Delete a record from the database.
   *
   * @param table The name of the table
   * @param key   The record key of the record to delete.
   * @return Zero on success, a non-zero error code on error
   */
  @Override
  public Status delete(String table, String key) {
    try {
      cache.remove(key);
      return Status.OK;
    } catch (Exception e) {
      log.error(String.format("Error deleting key: %s ", key), e);
    }

    return Status.ERROR;
  }

  /**
   * Entry processor to update values.
   */
  public static class Updater implements CacheEntryProcessor<String, BinaryObject, Object> {
    private String[] flds;
    private String[] vals;

    /**
     * @param values Updated fields.
     */
    Updater(Map<String, ByteIterator> values) {
      flds = new String[values.size()];
      vals = new String[values.size()];

      int idx = 0;
      for (Map.Entry<String, ByteIterator> e : values.entrySet()) {
        flds[idx] = e.getKey();
        vals[idx] = e.getValue().toString();
        ++idx;
      }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public Object process(MutableEntry<String, BinaryObject> mutableEntry, Object... objects)
              throws EntryProcessorException {
      BinaryObjectBuilder bob = mutableEntry.getValue().toBuilder();

      for (int i = 0; i < flds.length; ++i) {
        bob.setField(flds[i], vals[i]);
      }

      mutableEntry.setValue(bob.build());

      return null;
    }
  }
}

    try {
      BinaryObjectBuilder bob = cluster.binary().builder("CustomType");

      for (Map.Entry<String, ByteIterator> entry : values.entrySet()) {
        bob.setField(entry.getKey(), entry.getValue().toString());

        if (debug) {
          log.info(entry.getKey() + ":" + entry.getValue());
        }
      }

      BinaryObject bo = bob.build();

      if (table.equals(DEFAULT_CACHE_NAME)) {
        cache.put(key, bo);
      } else {
        throw new UnsupportedOperationException("Unexpected table name: " + table);
      }

      return Status.OK;
    } catch (Exception e) {
      log.error(String.format("Error inserting key: %s", key), e);

      return Status.ERROR;
    }
  
<methods>public non-sealed void <init>() ,public void cleanup() throws site.ycsb.DBException,public void init() throws site.ycsb.DBException,public site.ycsb.Status scan(java.lang.String, java.lang.String, int, Set<java.lang.String>, Vector<HashMap<java.lang.String,site.ycsb.ByteIterator>>) <variables>protected static final java.lang.String CLIENT_NODE_NAME,protected static final java.lang.String DEFAULT_CACHE_NAME,protected static final java.lang.String HOSTS_PROPERTY,protected static final java.util.concurrent.atomic.AtomicInteger INIT_COUNT,protected static final java.lang.String PORTS_DEFAULTS,protected static final java.lang.String PORTS_PROPERTY,protected static IgniteCache<java.lang.String,BinaryObject> cache,protected static Ignite cluster,protected static boolean debug,protected static TcpDiscoveryIpFinder ipFinder,protected static Logger log