title: Hazelcast Write Behind using Cassandra
body: <p>I am using Cassandra as a persistent store along with Hazelcast 3.6.For a rest API(POST). The Mapstore Implementation is as shown below.</p>
  
  <pre><code>import java.util.Collection;
  import java.util.HashMap;
  import java.util.Map;
  import java.util.concurrent.ExecutionException;
  
  import com.datastax.driver.core.Cluster;
  import com.datastax.driver.core.ResultSet;
  import com.datastax.driver.core.ResultSetFuture;
  import com.datastax.driver.core.Row;
  import com.datastax.driver.core.Session;
  import com.datastax.driver.core.querybuilder.Insert;
  import com.datastax.driver.core.querybuilder.QueryBuilder;
  import com.hazelcast.core.MapStore;
  
  public class InventoryMapStoreImpl implements MapStore&lt;String, Map&lt;String, Integer&gt;&gt; {
  
      public static Cluster cluster;
      public static Session session;
  
      public InventoryMapStoreImpl() {
          cluster = Cluster.builder().addContactPoint("127.0.0.1").build();
          session = cluster.connect("company");
      }
  
      @Override
      public synchronized Map&lt;String, Integer&gt;  load(String key) {
  
          Map&lt;String, Integer&gt; qty = new HashMap&lt;String, Integer&gt;();
          ResultSetFuture futureList = session
                  .executeAsync("Select * from company.onhandinventoryavailability WHERE key='" + key + "';");
          try {
  
              if (futureList.get() != null) {
                  ResultSet rsFuture = futureList.get();
                  Row resultOne = rsFuture.one();
                  if (resultOne != null) {
                      qty = resultOne.getMap(1, String.class, Integer.class);
                  }
              }
              session.close();
              cluster.close();
          } catch (InterruptedException e) {
              e.printStackTrace();
          } catch (ExecutionException e) {
              e.printStackTrace();
          }
          return qty;
      }
  
      @Override
      public Map&lt;String, Map&lt;String, Integer&gt;&gt; loadAll(Collection&lt;String&gt; keys) {
          return null;
      }
  
      @Override
      public Iterable&lt;String&gt; loadAllKeys() {
          return null;
      }
  
      @Override
      public synchronized void store(String key, Map&lt;String, Integer&gt; value) {
  
          try {
  
              Insert insert = QueryBuilder.insertInto("onhandinventoryavailability")
                      .value("key", key);
                insert.value("value", value);
                session.execute(insert);
              System.out.println("INSERTED:"+key+" INTO CASSANDRA...");
          } catch (Exception e) {
              System.out.println("ERORRRR");
              e.printStackTrace();
          }
          session.close();
          cluster.close();
      }
  
      @Override
      public void storeAll(Map&lt;String, Map&lt;String, Integer&gt;&gt; map) {
      }
  
      @Override
      public void delete(String key) {
      }
  
      @Override
      public void deleteAll(Collection&lt;String&gt; keys) {
      }
  }
  </code></pre>
  
  <p>And the hazelcast xml entry for the map is.</p>
  
  <pre><code>map name="onHandInventoryAvailability"&gt;
                  &lt;map-store enabled="true"&gt;
                          &lt;class-name&gt;com.company.common.hazelcast.mapstore.InventoryMapStoreImpl&lt;/class-name&gt;
                          &lt;write-delay-seconds&gt;5&lt;/write-delay-seconds&gt;
                          &lt;write-batch-size&gt;1000&lt;/write-batch-size&gt;
                          &lt;write-coalescing&gt;true&lt;/write-coalescing&gt;
                  &lt;/map-store&gt;
                  &lt;in-memory-format&gt;BINARY&lt;/in-memory-format&gt;
                  &lt;backup-count&gt;1&lt;/backup-count&gt;
                  &lt;async-backup-count&gt;0&lt;/async-backup-count&gt;
                  &lt;time-to-live-seconds&gt;0&lt;/time-to-live-seconds&gt;
                  &lt;max-idle-seconds&gt;0&lt;/max-idle-seconds&gt;
                  &lt;eviction-policy&gt;NONE&lt;/eviction-policy&gt;
                  &lt;max-size policy="PER_NODE"&gt;0&lt;/max-size&gt;
                  &lt;eviction-percentage&gt;25&lt;/eviction-percentage&gt;
                  &lt;min-eviction-check-millis&gt;100&lt;/min-eviction-check-millis&gt;
                  &lt;merge-policy&gt;com.hazelcast.map.merge.PutIfAbsentMapMergePolicy&lt;/merge-policy&gt;
                  &lt;cache-deserialized-values&gt;INDEX-ONLY&lt;/cache-deserialized-values&gt;
           &lt;/map&gt;
  </code></pre>
  
  <p>The problem is that when I hit the POST api after a delay or one at a time the data comes in both hazelcast and cassandra, But when I do more POSTs say 10,100 etc the connection to cassandra is lost by hazelcast and the write behind doen't work properly.Only one or two records out of 10,100 etc go cassandra(Not all).Is there anything wrong in my mapstore implementation?</p>