Duplicate entries when start new member of Hazelcast cluster

I implement my MapStore to save entries to database (external datastore), but there a issue of starting new node when old member storing entries to database. Firstly, cluster A has only Member1.

Client puts 20K entries to the cluster then Member1 starts saving entries by batch to database (write-behind mode). When 5K entries has been saved to database (not finish whole 20K entries) I start Member2 to join the cluster A

--> Member2 received about 10K entries (result of repartition) , Member2 stores them to the database, despite of this 10K entries has been saved to database by Member1

This leads to duplicate 10K records in database (10K from Member2). This is not occurs when 2 members are both ready when start putting entries from client, or in write-though mode

Please tell me why? This is my code

/**
 *
 * @author Mina Mimi
 */
public class SmsLogMapStore implements MapStore<String, SmsLog>, MapLoaderLifecycleSupport {

    Connection conn;

    static final Logger logger = Logger.getLogger(SmsLogMapStore.class.getSimpleName());

    @Override
    public synchronized void store(String k, SmsLog v) {
        logger.info("===write one >>>:" + v.getContent());
        String sql = "insert into smslog (content) values (?)";
        PreparedStatement st = null;
        try {
            st = conn.prepareStatement(sql);
            st.setString(1, v.getContent());
            st.executeUpdate();
        } catch (SQLException ex) {
            Logger.getLogger(SmsLogMapStore.class.getName()).log(Level.SEVERE, null, ex);
        } finally {
            try {
                st.close();
            } catch (SQLException ex) {
                Logger.getLogger(SmsLogMapStore.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
        logger.info("Finished write one >>>");
    }
    @Override
    public synchronized void storeAll(Map<String, SmsLog> map) {
        logger.info("Write batch:" + map.size());
        String sql = "insert into smslog (content) values (?)";
        PreparedStatement st = null;
        try {
            st = conn.prepareStatement(sql);
            for (Map.Entry<String, SmsLog> entry : map.entrySet()) {
                String key = entry.getKey();
                SmsLog v = entry.getValue();
                logger.info("===Writing(k,v):" + key + "," + v + " from batch");
                st.setString(1, v.getContent());
                st.addBatch();
            }
            st.executeBatch();
        } catch (SQLException ex) {
            Logger.getLogger(SmsLogMapStore.class.getName()).log(Level.SEVERE, null, ex);
        } finally {
            try {
                st.close();
            } catch (SQLException ex) {
                Logger.getLogger(SmsLogMapStore.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
        logger.info("Finished Write batch >>>");
    }

    @Override
    public void delete(String k) {
        return;
    }

    @Override
    public void deleteAll(Collection<String> clctn) {
        //throw new UnsupportedOperationException("Not supported yet."); //To change body of generated methods, choose Tools | Templates.
        for (String key : clctn) {
            delete(key);
        }

    }

    @Override
    public SmsLog load(String k) {        
        return new SmsLog();
    }

    @Override
    public Map<String, SmsLog> loadAll(Collection<String> clctn) {
        logger.info("############loadAll");
        HashMap<String, SmsLog> result = new HashMap<String, SmsLog>();
        return result;
    }

    @Override
    public synchronized Iterable<String> loadAllKeys() {
        List<String> keys = new LinkedList<String>();
        return keys;
    }

    @Override
    public void init(HazelcastInstance hi, Properties prprts, String string) {
        try {
            conn = getConnection();
        } catch (SQLException ex) {
            Logger.getLogger(SmsLogMapStore.class.getName()).log(Level.SEVERE, null, ex);
        }
    }

    @Override
    public void destroy() {
        try {
            conn.close();
        } catch (SQLException ex) {
            Logger.getLogger(SmsLogMapStore.class.getName()).log(Level.SEVERE, null, ex);
        }
    }

    private Connection getConnection() throws SQLException {
        Connection conn = DriverManager.getConnection("jdbc:mysql://127.0.0.1:3306/test_imdb", "root", "");
        return conn;
    }

}