Hazelcast Documentation

Hazelcast distributed queue is an implementation of java.util.concurrent.BlockingQueue.

import com.hazelcast.core.Hazelcast;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

BlockingQueue<MyTask> q = Hazelcast.getQueue("tasks");
q.put(new MyTask());
MyTask task = q.take();

boolean offered = q.offer(new MyTask(), 10, TimeUnit.SECONDS);
task = q.poll (5, TimeUnit.SECONDS);
if (task != null) {
	//process task
}

If you have 10 million tasks in your "tasks" queue and you are running that code over 10 JVMs (or servers), then each server carries 1 million task objects (plus backups). FIFO ordering will apply to all queue operations cluster-wide. User objects (such as MyTask in the example above), that are (en/de)queued have to be Serializable. Maximum capacity per JVM and the TTL (Time to Live) for a queue can be configured as shown in the example below.

<hazelcast>
	...
    <queue name="tasks">
        <!--
            Maximum size of the queue. When a JVM's local queue size reaches the maximum,
            all put/offer operations will get blocked until the queue size
            of the JVM goes down below the maximum.
            Any integer between 0 and Integer.MAX_VALUE. 0 means Integer.MAX_VALUE. Default is 0.
        -->
        <max-size-per-jvm>10000</max-size-per-jvm>

        <!--
            Maximum number of seconds for each item to stay in the queue. Items that are
            not consumed in <time-to-live-seconds> will get automatically evicted from the queue.
            Any integer between 0 and Integer.MAX_VALUE. 0 means infinite. Default is 0.
        -->
        <time-to-live-seconds>0</time-to-live-seconds>
    </queue>
</hazelcast>

As of version 1.9.3, distributed queues are backed by distributed maps. Thus, queues can have custom backup counts and persistent storage. Hazelcast will generate cluster-wide unique id for each element in the queue. Sample configuration:

<hazelcast>
	...
    <queue name="tasks">
        <!--
            Maximum size of the queue. When a JVM's local queue size reaches the maximum,
            all put/offer operations will get blocked until the queue size
            of the JVM goes down below the maximum.
            Any integer between 0 and Integer.MAX_VALUE. 0 means Integer.MAX_VALUE. Default is 0.
        -->
        <max-size-per-jvm>10000</max-size-per-jvm>
        
        <!--
            Name of the map configuration that will be used for the backing distributed
            map for this queue.
        -->
        <backing-map-ref>queue-map</backing-map-ref>
    </queue>

    <map name="queue-map">

        <backup-count>1</backup-count>

        <map-store enabled="true">

            <class-name>com.your,company.storage.DBMapStore</class-name>

            <write-delay-seconds>0</write-delay-seconds>

        </map-store>

        ...

    </map>
</hazelcast>

If the backing map has no map-store defined then your distributed queue will be in-memory only. If the backing map has a map-store defined then Hazelcast will call your MapStore implementation to persist queue elements. Even if you reboot your cluster Hazelcast will rebuild the queue with its content. When implementing a MapStore for the backing map, note that type of the key is always Long and values are the elements you place into the queue. So make sure MapStore.loadAllKeys returns Set<Long> for instance.

To learn about wildcard configuration feature, see Wildcard Configuration page.

Hazelcast provides distribution mechanism for publishing messages that are delivered to multiple subscribers which is also known as publish/subscribe (pub/sub) messaging model. Publish and subscriptions are cluster-wide. When a member subscribes for a topic, it is actually registering for messages published by any member in the cluster, including the new members joined after you added the listener. Messages are ordered, meaning, listeners(subscribers) will process the messages in the order they are actually published. If cluster member M publishes messages m1, m2, m3...mn to a topic T, then Hazelcast makes sure that all of the subscribers of topic T will receive and process m1, m2, m3...mn in order.

import com.hazelcast.core.Topic;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.MessageListener;

public class Sample implements MessageListener {

    public static void main(String[] args) { 
        Sample sample = new Sample();
        Topic topic = Hazelcast.getTopic ("default");  
        topic.addMessageListener(sample); 		
        topic.publish ("my-message-object");
    }  
	 
    public void onMessage(Object msg) {
        System.out.println("Message received = " + msg);
    } 
}

To learn about wildcard configuration feature, see Wildcard Configuration page.

Just like queue and set, Hazelcast will partition your map entries; and almost evenly distribute onto all Hazelcast members. Distributed maps have 1 backup (replica-count) by default so that if a member goes down, we don't lose data. Backup operations are synchronous so when a map.put(key, value) returns, it is guaranteed that the entry is replicated to one other node. For the reads, it is also guaranteed that map.get(key) returns the latest value of the entry. Consistency is strictly enforced.

import com.hazelcast.core.Hazelcast;
import java.util.Map;
import java.util.Collection;

Map<String, Customer> mapCustomers = Hazelcast.getMap("customers");
mapCustomers.put("1", new Customer("Joe", "Smith"));
mapCustomers.put("2", new Customer("Ali", "Selam"));
mapCustomers.put("3", new Customer("Avi", "Noyan"));

Collection<Customer> colCustomers = mapCustomers.values();
for (Customer customer : colCustomers) {
	// process customer
}

Hazelcast.getMap() actually returns com.hazelcast.core.IMap which extends java.util.concurrent.ConcurrentMap interface. So methods like ConcurrentMap.putIfAbsent(key,value) and ConcurrentMap.replace(key,value) can be used on distributed map as shown in the example below.

import com.hazelcast.core.Hazelcast;
import java.util.concurrent.ConcurrentMap;

Customer getCustomer (String id) {
    ConcurrentMap<String, Customer> map = Hazelcast.getMap("customers");
    Customer customer = map.get(id);
    if (customer == null) {
        customer = new Customer (id);
        customer = map.putIfAbsent(id, customer);
    }
    return customer;
}               

public boolean updateCustomer (Customer customer) {
    ConcurrentMap<String, Customer> map = Hazelcast.getMap("customers");
    return (map.replace(customer.getId(), customer) != null);            
}
                
public boolean removeCustomer (Customer customer) {
    ConcurrentMap<String, Customer> map = Hazelcast.getMap("customers");
    return map.remove(customer.getId(), customer) );           
}                                  
    	

All ConcurrentMap operations such as put and remove might wait if the key is locked by another thread in the local or remote JVM, but they will eventually return with success. ConcurrentMap operations never throwjava.util.ConcurrentModificationException.

Also see:

<hazelcast>
    ...
    <map name="default">
        <!--
            Number of backups. If 1 is set as the backup-count for example,
            then all entries of the map will be copied to another JVM for
            fail-safety. Valid numbers are 0 (no backup), 1, 2, 3.
        -->
        <backup-count>1</backup-count>

        <!--
            Can we read the local backup entries? Default value is false for
            strong consistency. Being able to read backup data will give you
            greater performance.
        -->
        <read-backup-data>false</read-backup-data>

        ...

    </map>
</hazelcast>

<hazelcast>
    ...
    <map name="default">
        <!--
            Number of backups. If 1 is set as the backup-count for example,
            then all entries of the map will be copied to another JVM for
            fail-safety. Valid numbers are 0 (no backup), 1, 2, 3.
        -->
        <backup-count>1</backup-count>

        <!--
            Maximum number of seconds for each entry to stay in the map. Entries that are
            older than <time-to-live-seconds> and not updated for <time-to-live-seconds>
            will get automatically evicted from the map.
            Any integer between 0 and Integer.MAX_VALUE. 0 means infinite. Default is 0.
        -->
        <time-to-live-seconds>0</time-to-live-seconds>

        <!--
            Maximum number of seconds for each entry to stay idle in the map. Entries that are
            idle(not touched) for more than <max-idle-seconds> will get
            automatically evicted from the map.
            Entry is touched if get, put or containsKey is called.
            Any integer between 0 and Integer.MAX_VALUE.
            0 means infinite. Default is 0.
        -->
        <max-idle-seconds>0</max-idle-seconds>

        <!--
            Valid values are:
            NONE (no extra eviction, <time-to-live-seconds> may still apply),
            LRU  (Least Recently Used),
            LFU  (Least Frequently Used).
            NONE is the default.
            Regardless of the eviction policy used, <time-to-live-seconds> will still apply. 
        -->
        <eviction-policy>LRU</eviction-policy>

        <!--
            Maximum size of the map. When max size is reached,
            map is evicted based on the policy defined.
            Any integer between 0 and Integer.MAX_VALUE. 0 means
            Integer.MAX_VALUE. Default is 0.
        -->
        <max-size>5000</max-size>

        <!--
            When max. size is reached, specified percentage of
            the map will be evicted. Any integer between 0 and 100.
            If 25 is set for example, 25% of the entries will
            get evicted.
        -->
        <eviction-percentage>25</eviction-percentage>
       <!--
            Specifies when eviction will be started. Default value is 3. 
           So every 3 (+up to 5 for performance reasons) seconds 
           eviction will be kicked of. Eviction is costly operation, setting 
           this number too low, can decrease the performance. 
       -->
      <eviction-delay-seconds>3</eviction-delay-seconds>
    </map>
</hazelcast>

<hazelcast>
    ...
    <map name="default">
        ...
        <map-store enabled="true">
            <!--
               Name of the class implementing MapLoader and/or MapStore. 
               The class should implement at least of these interfaces and
               contain no-argument constructor. Note that the inner classes are not supported.
            -->
            <class-name>com.hazelcast.examples.DummyStore</class-name>
            <!--
               Number of seconds to delay to call the MapStore.store(key, value).
               If the value is zero then it is write-through so MapStore.store(key, value)
               will be called as soon as the entry is updated.
               Otherwise it is write-behind so updates will be stored after write-delay-seconds
               value by calling Hazelcast.storeAll(map). Default value is 0.
            -->
            <write-delay-seconds>0</write-delay-seconds>
        </map-store>
    </map>
</hazelcast>

import java.io.Serializable;

public class Employee implements Serializable {
        private String name;
        private int age;
        private boolean active;
        private double salary;

        public Employee(String name, int age, boolean live, double price) {
            this.name = name;
            this.age = age;
            this.active = live;
            this.salary = price;
        }

        public Employee() {
        }

        public String getName() {
            return name;
        }

        public int getAge() {
            return age;
        }

        public double getSalary() {
            return salary;
        }

        public boolean isActive() {
            return active;
        }
}

import com.hazelcast.core.IMap;
import com.hazelcast.query.SqlPredicate;

IMap map = Hazelcast.getMap("employee");

Set<Employee> employees = (Set<Employee>) map.values(new SqlPredicate("active AND age < 30"));

import com.hazelcast.core.IMap;
import com.hazelcast.query.Predicate;
import com.hazelcast.query.PredicateBuilder;
import com.hazelcast.query.EntryObject;

IMap map = Hazelcast.getMap("employee");

EntryObject e = new PredicateBuilder().getEntryObject();
Predicate predicate = e.is("active").and(e.get("age").lessThan(30));

Set<Employee> employees = (Set<Employee>) map.values(predicate);

IMap imap = Hazelcast.getMap("employees");
imap.addIndex("age", true);        // ordered, since we have ranged queries for this field
imap.addIndex("active", false);    // not ordered, because boolean field cannot have range

<hazelcast>
    ...
    <map name="my-read-mostly-map">
        ...
        <near-cache>
            <!--
                Maximum number of seconds for each entry to stay in the near cache. Entries that are
                older than <time-to-live-seconds> will get automatically evicted from the near cache.
                Any integer between 0 and Integer.MAX_VALUE. 0 means infinite. Default is 0.
            -->
            <time-to-live-seconds>0</time-to-live-seconds>

            <!--
                Maximum number of seconds each entry can stay in the near cache as untouched (not-read).
                Entries that are not read (touched) more than <max-idle-seconds> value will get removed
                from the near cache.
                Any integer between 0 and Integer.MAX_VALUE. 0 means
                Integer.MAX_VALUE. Default is 0.
            -->
            <max-idle-seconds>60</max-idle-seconds>

            <!--
                Valid values are:
                NONE (no extra eviction, <time-to-live-seconds> may still apply),
                LRU  (Least Recently Used),
                LFU  (Least Frequently Used).
                NONE is the default.
                Regardless of the eviction policy used, <time-to-live-seconds> will still apply.
            -->
            <eviction-policy>LRU</eviction-policy>

            <!--
                Maximum size of the near cache. When max size is reached,
                cache is evicted based on the policy defined.
                Any integer between 0 and Integer.MAX_VALUE. 0 means
                Integer.MAX_VALUE. Default is 0.
            -->
            <max-size>5000</max-size>

            <!--
                Should the cached entries get evicted if the entries are changed (updated or removed).
                true of false. Default is true.
            -->
            <invalidate-on-change>true</invalidate-on-change>

        </near-cache>
    </map>
</hazelcast>

import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.MapEntry;

MapEntry entry = Hazelcast.getMap("quotes").getMapEntry("1");
System.out.println ("size in memory  : " + entry.getCost();
System.out.println ("creationTime    : " + entry.getCreationTime();
System.out.println ("expirationTime  : " + entry.getExpirationTime();
System.out.println ("number of hits  : " + entry.getHits();
System.out.println ("lastAccessedTime: " + entry.getLastAccessTime();
System.out.println ("lastUpdateTime  : " + entry.getLastUpdateTime();
System.out.println ("version         : " + entry.getVersion();
System.out.println ("isValid         : " + entry.isValid();
System.out.println ("key             : " + entry.getKey();
System.out.println ("value           : " + entry.getValue();
System.out.println ("oldValue        : " + entry.setValue(newValue);

To learn about wildcard configuration feature, see Wildcard Configuration page.

MultiMap is a specialized map where you can associate a key with multiple values. Just like any other distributed data structure implementation in Hazelcast, MultiMap is distributed/partitioned and thread-safe.

import com.hazelcast.core.MultiMap;
import com.hazelcast.core.Hazelcast;
import java.util.Collection;

// a multimap to hold <customerId, Order> pairs
MultiMap<String, Order> mmCustomerOrders = Hazelcast.getMultiMap("customerOrders");
mmCustomerOrders.put("1", new Order ("iPhone", 340));
mmCustomerOrders.put("1", new Order ("MacBook", 1200));
mmCustomerOrders.put("1", new Order ("iPod", 79));

// get orders of the customer with customerId 1.
Collection<Order> colOrders = mmCustomerOrders.get ("1");
for (Order order : colOrders) {
	// process order
}

// remove specific key/value pair
boolean removed = mmCustomerOrders.remove("1", new Order ("iPhone", 340));

Distributed Set is distributed and concurrent implementation of java.util.Set. Set doesn't allow duplicate elements, so elements in the set should have proper hashCode and equals methods.

import com.hazelcast.core.Hazelcast;
import java.util.Set;
import java.util.Iterator;

java.util.Set set = Hazelcast.getSet("IBM-Quote-History");
set.add(new Price(10, time1));
set.add(new Price(11, time2));
set.add(new Price(12, time3));
set.add(new Price(11, time4));
//....
Iterator it = set.iterator();
while (it.hasNext()) { 
    Price price = (Price) it.next(); 
    //analyze
}

Distributed List is very similar to distributed set, but it allows duplicate elements.

import com.hazelcast.core.Hazelcast;
import java.util.List;
import java.util.Iterator;

java.util.List list = Hazelcast.getList("IBM-Quote-Frequency");
list.add(new Price(10));
list.add(new Price(11));
list.add(new Price(12));
list.add(new Price(11));
list.add(new Price(12));
        
//....
Iterator it = list.iterator();
while (it.hasNext()) { 
    Price price = (Price) it.next(); 
    //analyze
}

import com.hazelcast.core.Hazelcast;
import java.util.concurrent.locks.Lock;

Lock lock = Hazelcast.getLock(myLockedObject);
lock.lock();
try {
    // do something here
} finally {
    lock.unlock();
} 
 

java.util.concurrent.locks.Lock.tryLock() with timeout is also supported. All operations on the Lock that Hazelcast.getLock(Object obj) returns are cluster-wide and Lock behaves just like java.util.concurrent.lock.ReentrantLock.

if (lock.tryLock (5000, TimeUnit.MILLISECONDS)) {
    try {  
       // do some stuff here..  
    } 
    finally {  
      lock.unlock();  
    }   
} 

Locks are fail-safe. If a member holds a lock and some of the members go down, cluster will keep your locks safe and available. Moreover, when a member leaves the cluster, all the locks acquired by this dead member will be removed so that these locks can be available for live members immediately.

Hazelcast allows you to register for entry events to get notified when entries added, updated or removed. Listeners are cluster-wide. When a member adds a listener, it is actually registering for events originated in any member in the cluster. When a new member joins, events originated at the new member will also be delivered. All events are ordered, meaning, listeners will receive and process the events in the order they are actually occurred.

import java.util.Queue;
import java.util.Map; 
import java.util.Set; 
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.ItemListener;
import com.hazelcast.core.EntryListener;
import com.hazelcast.core.EntryEvent; 

public class Sample implements ItemListener, EntryListener {

	public static void main(String[] args) { 
		Sample sample = new Sample();
		Queue queue = Hazelcast.getQueue ("default"); 
		Map   map   = Hazelcast.getMap   ("default"); 
		Set   set   = Hazelcast.getSet   ("default"); 
		//listen for all added/updated/removed entries
		queue.addItemListener(sample, true);
		set.addItemListener  (sample, true); 
		map.addEntryListener (sample, true);		
		//listen for an entry with specific key 
		map.addEntryListener (sample, "keyobj");		
	} 

	public void entryAdded(EntryEvent event) {
		System.out.println("Entry added key=" + event.getKey() + ", value=" + event.getValue());
	}

	public void entryRemoved(EntryEvent event) {
		System.out.println("Entry removed key=" + event.getKey() + ", value=" + event.getValue());
	}

	public void entryUpdated(EntryEvent event) {
		System.out.println("Entry update key=" + event.getKey() + ", value=" + event.getValue());
	} 

	public void entryEvicted(EntryEvent event) {
		System.out.println("Entry evicted key=" + event.getKey() + ", value=" + event.getValue());
	} 
	
	public void itemAdded(Object item) {
		System.out.println("Item added = " + item);
	}

	public void itemRemoved(Object item) {
		System.out.println("Item removed = " + item);
	}	 
}
       

Hazelcast allows you to register for membership events to get notified when members added or removed. You can also get the set of cluster members.

import com.hazelcast.core.*;

Cluster cluster = Hazelcast.getCluster();
cluster.addMembershipListener(new MembershipListener(){
	public void memberAdded(MembershipEvent membersipEvent) {
		System.out.println("MemberAdded " + membersipEvent);
	}

	public void memberRemoved(MembershipEvent membersipEvent) {
		System.out.println("MemberRemoved " + membersipEvent);
	}
});

Member localMember  = cluster.getLocalMember();
System.out.println ("my inetAddress= " + localMember.getInetAddress());

Set setMembers  = cluster.getMembers();
for (Member member : setMembers) {
	System.out.println ("isLocalMember " + member.localMember());
	System.out.println ("member.inetaddress " + member.getInetAddress());
	System.out.println ("member.port " + member.getPort());
}

Hazelcast IdGenerator creates cluster-wide unique IDs. Generated IDs are long type primitive values between 0 and Long.MAX_VALUE . Id generation occurs almost at the speed of AtomicLong.incrementAndGet() . Generated IDs are unique during the life cycle of the cluster. If the entire cluster is restarted, IDs start from 0 again.

import com.hazelcast.core.IdGenerator;
import com.hazelcast.core.Hazelcast;

IdGenerator idGenerator = Hazelcast.getIdGenerator("customer-ids");
long id = idGenerator.newId();

Super Clients are members with no storage. If -Dhazelcast.super.client=true JVM parameter is set, then the JVM will join the cluster as a 'super client' which will not be a 'data partition' (no data on that node) but will have super fast access to the cluster just like any regular member does.

Hazelcast can be used in transactional context. Basically start a transaction, work with queues, maps, sets and do other things then commit/rollback in one shot.

import java.util.Queue;
import java.util.Map;
import java.util.Set;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.Transaction; 

Queue queue = Hazelcast.getQueue("myqueue");
Map map     = Hazelcast.getMap  ("mymap");
Set set     = Hazelcast.getSet  ("myset");

Transaction txn = Hazelcast.getTransaction();
txn.begin();
try {	
	Object obj = queue.poll();
	//process obj
	map.put ("1", "value1");
	set.add ("value");
	//do other things..
	txn.commit();
}catch (Throwable t)  {
	txn.rollback();
}

Isolation is always READ_COMMITTED . If you are in a transaction, you can read the data in your transaction and the data that is already committed and if not in a transaction, you can only read the committed data. Implementation is different for queue and map/set. For queue operations (offer,poll), offered and/or polled objects are copied to the next member in order to safely commit/rollback. For map/set, Hazelcast first acquires the locks for the write operations (put, remove) and holds the differences (what is added/removed/updated) locally for each transaction. When transaction is set to commit, Hazelcast will release the locks and apply the differences. When rolling back, Hazelcast will simply releases the locks and discard the differences. Transaction instance is attached to the current thread and each Hazelcast operation checks if the current thread holds a transaction, if so, operation will be transaction aware. When transaction is committed, rolled back or timed out, it will be detached from the thread holding it.

Hazelcast can be integrated into J2EE containers via Hazelcast Resource Adapter ( hazelcast-ra.rar ). After proper configuration, Hazelcast can participate in standard J2EE transactions.

<%@page import="javax.resource.ResourceException" %>
<%@page import="javax.transaction.*" %>
<%@page import="javax.naming.*" %>
<%@page import="javax.resource.cci.*" %> 
<%@page import="java.util.*" %> 
<%@page import="com.hazelcast.core.Hazelcast" %> 

<%
UserTransaction txn = null;
Connection conn = null;
Queue queue = Hazelcast.getQueue ("default");
Map map     = Hazelcast.getMap   ("default");
Set set     = Hazelcast.getSet   ("default");
List list   = Hazelcast.getList  ("default");

try { 
    Context context = new InitialContext(); 
    txn = (UserTransaction) context.lookup("java:comp/UserTransaction"); 
    txn.begin(); 
	
    ConnectionFactory cf = (ConnectionFactory) context.lookup ("java:comp/env/HazelcastCF"); 
    conn = cf.getConnection();  
	
    queue.offer("newitem");
    map.put ("1", "value1");
    set.add ("item1");
    list.add ("listitem1");
	
    txn.commit(); 
} catch (Throwable e) { 
    if (txn != null) {
        try {
            txn.rollback();
        } catch (Exception ix) {ix.printStackTrace();};
    }
    e.printStackTrace();
} finally { 
    if (conn != null) {
        try {
            conn.close();
        } catch (Exception ignored) {};
    }
}
%>

Distributed executor service is a distributed implementation of java.util.concurrent.ExecutorService. It allows you to execute your code in cluster. In this chapter, all the code samples are based on the Echo class above. Please note that Echo class is Serializable . You can ask Hazelcast to execute your code (Runnable, Callable):

import com.hazelcast.core.Member;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.MultiTask;
import com.hazelcast.core.DistributedTask;
import java.util.concurrent.ExecutorService;   
import java.util.concurrent.FutureTask;   
import java.util.concurrent.Future;   
import java.util.Set;

public void echoOnTheMember(String input, Member member) throws Exception {
       FutureTask<String> task = new DistributedTask<String>(new Echo(input), member);
       ExecutorService executorService = Hazelcast.getExecutorService();
       executorService.execute(task);
       String echoResult = task.get();
}

public void echoOnTheMemberOwningTheKey(String input, Object key) throws Exception {
       FutureTask<String> task = new DistributedTask<String>(new Echo(input), key);
       ExecutorService executorService = Hazelcast.getExecutorService();
       executorService.execute(task);
       String echoResult = task.get();
}

public void echoOnSomewhere(String input) throws Exception { 
       ExecutorService executorService = Hazelcast.getExecutorService();
       Future<String> task = executorService.submit(new Echo(input));
       String echoResult = task.get();
}

public void echoOnMembers(String input, Set<Member> members) throws Exception {
       MultiTask<String> task = new MultiTask<String>(new Echo(input), members);
       ExecutorService executorService = Hazelcast.getExecutorService();
       executorService.execute(task);
       Collection<String> results = task.get(); 
} 

Note that you can obtain the set of cluster members via Hazelcast.getCluster().getMembers() call. You can also extend the MultiTask class to override set(V result), setException(Throwable exception), done() methods for custom behaviour. Just like java.util.concurrent.FutureTask.get() , MultiTask.get() will throw java.util.concurrent.ExecutionException if any of the executions throws exception.

What if the code you execute in cluster takes longer than acceptable. If you cannot stop/cancel that task it will keep eating your resources. Standard Java executor framework solves this problem with by introducing cancel() api and 'encouraging' us to code and design for cancellations, which is highly ignored part of software development.

public class Fibonacci<Long> implements Callable<Long>, Serializable {
    int input = 0; 

    public Fibonacci() { 
    } 

    public Fibonacci(int input) { 
        this.input = input;
    } 

    public Long call() {
        return calculate (input);
    }

    private long calculate (int n) {
        if (Thread.currentThread().isInterrupted()) return 0;
        if (n <= 1) return n;
        else return calculate(n-1) + calculate(n-2);
    }
}

The callable class above calculates the fibonacci number for a given number. In the calculate method, we are checking to see if the current thread is interrupted so that code can be responsive to cancellations once the execution started. Following fib() method submits the Fibonacci calculation task for number 'n' and waits maximum 3 seconds for result. If the execution doesn't complete in 3 seconds, future.get() will throw TimeoutException and upon catching it we interruptibly cancel the execution for saving some CPU cycles.

long fib(int n) throws Exception {
    ExecutorService es = Hazelcast.getExecutorService();
    Future future = es.submit(new Fibonacci(n));  
    try {
        return future.get(3, TimeUnit.SECONDS);
    } catch (TimeoutException e) {
        future.cancel(true);            
    }
    return -1;
}

fib(20) will probably will take less than 3 seconds but fib(50) will take way longer. (This is not the example for writing better fibonacci calculation code but for showing how to cancel a running execution that takes too long.) future.cancel(false) can only cancel execution before it is running (executing) but future.cancel(true) can interrupt running executions if your code is able to handle the interruption. So if you are willing to be able to cancel already running task then your task has to be designed to handle interruption. If calculate (int n) method didn't have if (Thread.currentThread().isInterrupted()) line, then you wouldn't be able to cancel the execution after it started.

ExecutionCallback allows you to asynchronously get notified when the execution is done. When implementing ExecutionCallback.done(Future) method, you can check if the task is already cancelled.

import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.ExecutionCallback;
import com.hazelcast.core.DistributedTask;
import java.util.concurrent.ExecutorService;    
import java.util.concurrent.Future; 
		
ExecutorService es = Hazelcast.getExecutorService();
DistributedTask<String> task = new DistributedTask<String>(new Fibonacci<Long>(10));
task.setExecutionCallback(new ExecutionCallback<Long> () {
	public void done (Future<Long> future) { 
		try {
			if (! future.isCancelled()) {
                System.out.println("Fibonacci calculation result = " + future.get());
            }
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
});
es.execute(task);

You could have achieved the same results by extending DistributedTask and overriding the DistributedTask.done() method.

import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.DistributedTask;
import java.util.concurrent.ExecutorService;    
import java.util.concurrent.Future; 
		
ExecutorService es = Hazelcast.getExecutorService();
es.execute(new DistributedTask<String>(new Fibonacci<Long>(10)) {
	public void done () { 
		try {
            if (! isCancelled()) {
                System.out.println("Fibonacci calculation result = " + get());
            }
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}); 

If multicast is not prefered way of discovery for your environment, then you can configure Hazelcast for full TCP/IP cluster. As configuration below shows, while enable attribute of multicast is set to false, tcp-ip has to be set to true. For the none-multicast option, all or subset of cluster members' hostnames and/or ip addreses must be listed. Note that all of the cluster members don't have to be listed there but at least one of them has to be active in cluster when a new member joins.

<hazelcast>
	...
	<network>
		<port auto-increment="true">5701</port>
		<join>
			<multicast enabled="false">
				<multicast-group>224.2.2.3</multicast-group>
				<multicast-port>54327</multicast-port>
			</multicast>
			<tcp-ip enabled="true">
				<hostname>machine1</hostname>
				<hostname>machine2</hostname>
				<hostname>machine3:5799</hostname>
				<interface>192.168.1.0-7</interface> 	
				<interface>192.168.1.21</interface> 
			</tcp-ip>
		</join>
		...
	</network>
	...
</hazelcast>

Hazelcast supports EC2 Auto Discovery as of 1.9.4. It is useful when you don't want or can't provide the list of possible IP addresses. Here is a sample configuration: Disable join over multicast and tcp/ip and enable aws. Also provide the credentials.

<join>
    <multicast enabled="false">
        <multicast-group>224.2.2.3</multicast-group>
        <multicast-port>54327</multicast-port>
    </multicast>
    <tcp-ip enabled="false">
        <interface>192.168.1.2</interface>
    </tcp-ip>
    <aws enabled="true">
        <access-key>my-access-key</access-key>
        <secret-key>my-secret-key</secret-key>
        <region>us-west-1</region>                              <!-- optional, default is us-east-1 -->
        <security-group-name>hazelcast-sg</security-group-name> <!-- optional -->
        <tag-key>type</tag-key>                                 <!-- optional -->
        <tag-value>hz-nodes</tag-value>                         <!-- optional -->
    </aws>
</join>

You need to add hazelcast-cloud.jar dependency into your project. Note that it is also bundled inside hazelcast-all.jar. hazelcast-cloud module doesn't depend on any other third party modules.

By specifying group-name and group-password, you can separate your clusters in a simple way; dev group, production group, test group, app-a group etc...

<hazelcast>
	<group>
		<name>dev</name>
		<password>dev-pass</password>
	</group>
	...
	
</hazelcast>

You can also set the groupName with Config API. JVM can host multiple Hazelcast instances (nodes). Each node can only participate in one group and it only joins to its own group, does not mess with others. Following code creates 3 separate Hazelcast nodes, h1 belongs to app1 cluster, while h2 and h3 are belong to app2 cluster.

<hazelcast>
        Config configApp1 = new Config();
        configApp1.getGroupConfig().setName("app1");

        Config configApp2 = new Config();
        configApp2.getGroupConfig().setName("app2");

        HazelcastInstance h1 = Hazelcast.newHazelcastInstance(configApp1);
        HazelcastInstance h2 = Hazelcast.newHazelcastInstance(configApp2);
        HazelcastInstance h3 = Hazelcast.newHazelcastInstance(configApp2);

You can also specify which network interfaces that Hazelcast should use. Servers mostly have more than one network interface so you may want to list the valid IPs. Range characters ('*' and '-') can be used for simplicity. So 10.3.10.*, for instance, refers to IPs between 10.3.10.0 and 10.3.10.255. Interface 10.3.10.4-18 refers to IPs between 10.3.10.4 and 10.3.10.18 (4 and 18 included). If network interface configuration is enabled (disabled by default) and if Hazelcast cannot find an matching interface, then it will print a message on console and won't start on that node.

<hazelcast>
	...
	<network>
		....
		<interfaces enabled="true">
			<interface>10.3.16.*</interface> 
			<interface>10.3.10.4-18</interface> 
			<interface>192.168.1.3</interface> 		
		</interfaces>	
	</network>
	...
</hazelcast> 

Imagine that you have 10-node cluster and for some reason the network is divided into two in a way that 4 servers cannot see the other 6. As a result you ended up having two separate clusters; 4-node cluster and 6-node cluster. Members in each sub-cluster are thinking that the other nodes are dead even though they are not. This situation is called Network Partitioning (aka Split-Brain Syndrome).

Since it is a network failure, there is no way to avoid it programatically and your application will run as two separate independent clusters but we should be able answer the following questions: "What will happen after the network failure is fixed and connectivity is restored between these two clusters? Will these two clusters merge into one again? If they do, how are the data conflicts resolved, because you might end up having two different values for the same key in the same map?"

Here is how Hazelcast deals with it:

So each member of the merging cluster is actually rejoining to the new cluster and sending merge request for each its locally owned map entry.

Q: Which cluster will merge into the other?

A. Smaller cluster will merge into the bigger one. If they have equal number of members then a hashing algorithm determines the merging cluster.

Q. Each cluster may have different versions of the same key in the same map. How is the conflict resolved?

A. Destination cluster will decide how to handle merging entry based on the MergePolicy set for that map. There are built-in merge policies such as hz.NO_MERGE, hz.ADD_NEW_ENTRY and hz.LATEST_UPDATE but you can develop your own merge policy by implementingcom.hazelcast.merge.MergePolicy. You should register your custom merge policy in the configuration so that Hazelcast can find it by name.

   public interface MergePolicy {
       /**
        * Returns the value of the entry after the merge
        * of entries with the same key. Returning value can be
        * You should consider the case where existingEntry is null.
        *
        * @param mapName       name of the map
        * @param mergingEntry  entry merging into the destination cluster
        * @param existingEntry existing entry in the destination cluster
        * @return final value of the entry. If returns null then no change on the entry.
        */
       Object merge(String mapName, MapEntry mergingEntry, MapEntry existingEntry);
   }

Here is how merge policies are registered and specified per map.

<hazelcast>
    ...
    <map name="default">
        <backup-count>1</backup-count>
        <eviction-policy>NONE</eviction-policy>
        <max-size>0</max-size>
        <eviction-percentage>25</eviction-percentage>
        <!--
            While recovering from split-brain (network partitioning),
            map entries in the small cluster will merge into the bigger cluster
            based on the policy set here. When an entry merge into the
            cluster, there might an existing entry with the same key already.
            Values of these entries might be different for that same key.
            Which value should be set for the key? Conflict is resolved by
            the policy set here. Default policy is hz.ADD_NEW_ENTRY

            There are built-in merge policies such as
            hz.NO_MERGE      ; no entry will merge.
            hz.ADD_NEW_ENTRY ; entry will be added if the merging entry's key
                               doesn't exist in the cluster.
            hz.HIGHER_HITS   ; entry with the higher hits wins.
            hz.LATEST_UPDATE ; entry with the latest update wins.
        -->
        <merge-policy>MY_MERGE_POLICY</merge-policy>
    </map>

    <merge-policies>
        <map-merge-policy name="MY_MERGE_POLICY">
            <class-name>com.acme.MyOwnMergePolicy</class-name>
        </map-merge-policy>
    </merge-policies>
    ...
</hazelcast>

Hazelcast supports wildcard configuration of Maps, Queues and Topics. Using an asterisk (*) character in the name, different instances of Maps, Queues and Topics can be configured by a single configuration.

Note that, with a limitation of a single usage, asterisk (*) can be placed anywhere inside the configuration name.

For instance a map named 'com.hazelcast.test.mymap' can be configured using one of these configurations;

<map name="com.hazelcast.test.*">
...
</map>
        

<map name="com.hazel*">
...
</map>
        

<map name="*.test.mymap">
...
</map>
        

<map name="com.*test.mymap">
...
</map>
        

Or a queue 'com.hazelcast.test.myqueue'

<queue name="*hazelcast.test.myqueue">
...
</queue>
        

<queue name="com.hazelcast.*.myqueue">
...
</queue>
        

There are some advanced configuration properties to tune some aspects of Hazelcast. These can be set as property name and value pairs through configuration xml, configuration API or JVM system property.

Hazelcast has a flexible logging configuration and doesn't depend on any logging framework except JDK logging. It has in-built adaptors for a number of logging frameworks and also supports custom loggers by providing logging interfaces.

To use built-in adaptors you should set hazelcast.logging.type property to one of predefined types below.

You can set hazelcast.logging.type through configuration xml, configuration API or JVM system property.

To use custom logging feature you should implement com.hazelcast.logging.LoggerFactory and com.hazelcast.logging.ILogger interfaces and set system property hazelcast.logging.class to your custom LoggerFactory class name.

	java -Dhazelcast.logging.class=foo.bar.MyLoggingFactory

You can also listen logging events generated by Hazelcast runtime by registering LogListeners to LoggingService.

	LogListener listener = new LogListener() {
		public void log(LogEvent logEvent) {
			// do something
		}
	}
	
	LoggingService loggingService = Hazelcast.getLoggingService();
	loggingService.addLogListener(Level.INFO, listener):
	

Through the LoggingService you can get the current used ILogger implementation and log your own messages too.

Native Client enables you to do all Hazelcast operations without being a member of the cluster. It connects to one of the cluster members and delegates all cluster wide operations to it. When the relied cluster member dies, client will transparently switch to another live member.

There can be hundreds, even thousands of clients connected to the cluster.

Imagine a trading application where all the trading data stored and managed in a 10 node Hazelcast cluster. Swing/Web applications at traders' desktops can use Native Java Client to access and modify the data in the Hazelcast cluster.

Currently Hazelcast has Native Java Client available. The next client implementation will be CSharp.

Super Client vs. Native Client

Super Client is a member of the cluster, it has socket connection to every member in the cluster and it knows where the data is so it will get to the data much faster. But Super Client has the clustering overhead and it must be on the same data center even on the same RAC. However Native client is not member and relies on one of the cluster members. Native Clients can be anywhere in the LAN or WAN. It scales much better and overhead is quite less. So if your clients are less than Hazelcast nodes then Super client can be an option; otherwise definitely try Native Client. As a rule of thumb: Try Native client first, if it doesn't perform well enough for you, then consider Super client.

The following picture describes the clients connecting to Hazelcast Cluster. Note the difference between Super Client and Java Client

    import com.hazelcast.core.HazelcastInstance;
    import com.hazelcast.client.HazelcastClient;

    import java.util.Map;
    import java.util.Collection;


    HazelcastInstance client = HazelcastClient.newHazelcastClient("dev", "dev-pass", "10.90.0.1", "10.90.0.2:5702");
    //All Cluster Operations that you can do with ordinary HazelcastInstance
    Map<String, Customer> mapCustomers = client.getMap("customers");
    mapCustomers.put("1", new Customer("Joe", "Smith"));
    mapCustomers.put("2", new Customer("Ali", "Selam"));
    mapCustomers.put("3", new Customer("Avi", "Noyan"));

    Collection<Customer> colCustomers = mapCustomers.values();
    for (Customer customer : colCustomers) {
         // process customer
    }

A Memcache client written in any language can talk directly to Hazelcast cluster. No additional configuration is required. Here is an example: Let's say your cluster's members are:

Members [5] {
    Member [10.20.17.1:5701]
    Member [10.20.17.2:5701]
    Member [10.20.17.4:5701]
    Member [10.20.17.3:5701]
    Member [10.20.17.5:5701]
 }

And you have a PHP application that uses PHP Memcache client to cache things in Hazelcast. All you need to do is have your PHP memcache client connect to one of these members. It doesn't matter which member the client connects to because Hazelcast cluster looks as one giant machine (Single System Image). PHP client code sample:

<?php
    $memcache = new Memcache;
    $memcache->connect('10.20.17.1', 5701) or die ("Could not connect");
    $memcache->set('key1','value1',0,3600);
    $get_result = $memcache->get('key1'); //retrieve your data
    var_dump($get_result); //show it
?>

Notice that memcache client is connecting to 10.20.17.1 and using port 5701. Java client code sample with SpyMemcached client:

MemcachedClient client = new MemcachedClient(AddrUtil.getAddresses("10.20.17.1:5701 10.20.17.2:5701"));
client.set("key1", 3600, "value1");
System.out.println(client.get("key1"));

An entry written with a memcache client can be read by another memcache client written in another language.

Let's say your cluster's members are:

Members [5] {
    Member [10.20.17.1:5701]
    Member [10.20.17.2:5701]
    Member [10.20.17.4:5701]
    Member [10.20.17.3:5701]
    Member [10.20.17.5:5701]
 }

And you have a distributed map named 'stocks'. You can put a new key1/value1 entry into this map by issuing HTTP POST call to http://10.20.17.1:5701/hazelcast/rest/maps/stocks/key1 URL. Your http post call's content body should contain the value (value1). You can retrieve this entry via HTTP GET call to http://10.20.17.1:5701/hazelcast/rest/maps/stocks/key1. You can also retrieve this entry from another member such as http://10.20.17.3:5701/hazelcast/rest/maps/stocks/key1.

RESTful access is provided through any member of your cluster. So you can even put an HTTP load-balancer in-front of your cluster members for load-balancing and fault-tolerance.

Now go ahead and install a REST plugin for your browser and explore further.

Hazelcast also stores the mime-type of your POST request if it contains any. So if, for example, you post binary of an image file and set the mime-type of the HTTP POST request to image/jpeg then this mime-type will be part of the response of your HTTP GET request for that entry.

Let's say you also have a task queue named 'tasks'. You can offer a new item into the queue via HTTP POST and take and item from the queue via HTTP DELETE.

HTTP POST http://10.20.17.1:5701/hazelcast/rest/queues/tasks <CONTENT> means

Hazelcast.getQueue("tasks").offer(<CONTENT>);

and HTTP DELETE http://10.20.17.1:5701/hazelcast/rest/queues/tasks/3 means

Hazelcast.getQueue("tasks").poll(3, SECONDS);

Note that you will have to handle the failures on REST polls as there is no transactional guarantee.

In this section, we will go over the Hazelcast's internal threads, the client threads executing Hazelcast API and how these threads work together in Hazelcast. When developing Hazelcast, you should know which thread will execute your code, which variables are local to that thread, and how you should interact with other threads.

1. Client Threads:

Client threads are your threads, user's application threads, and or user's application/web server's threads that are executing Hazelcast API. User's threads that are client to Hazelcast. For example, Hazelcast.getQueue("myqueue"), map.put(key, value), set.size() calls are initiated and finalized in the client threads. Serialization of the objects also happens in the client threads. This also eliminates the problems associated with classloaders. Client threads initiate the calls, serialize the objects into Hazelcast com.hazelcast.nio.Data object which holds a java.nio.ByteBuffer. Data object is the binary representation of the application objects (key, value, item, callable objects). All Hazelcast threads such as ServiceThread, InThread and OutThread work with Data objects; they don't know anything about the actual application objects. When the calls are finalized, if the return type is an object, Data object is returned to the client thread and client thread then will deserialize the Data (binary representation) back to the application objects.

For each client thread, there is a com.hazelcast.impl.ThreadContext thread-local instance attached which contains thread context information such as transaction.

2. ServiceThread:

ServiceThread, implemented at com.hazelcast.impl.ClusterService, is the most significant thread in Hazelcast. Almost all none-IO operations happens in this thread. ServiceThread serves to local client calls and the calls from other members.

If you look at the ClusterService class you will see it is constantly dequeueing its queue and processing local and remote events. ClusterService queue receives local events in the form of com.hazelcast.impl.BaseManager.Processable instances and remote events in the form of com.hazelcast.nio.PacketQueue.Packet instances from InThread.

All distributed data structures (queue, map, set) are eventually modified in this thread so there is -no- synchronization needed when data structures are accessed/modified.

It is important to understand that client threads initiates/finalizes the calls, in/out threads handles the socket read/writes and ServiceThread does the actually manipulation of the data structures. There is no other threads allowed to touch the data structures (maps, queues) so that there is no need to protect the data structures from multithread access: no synchronization when accessing data structures. It may sound inefficient to allow only one thread to do all data structure updates but here is the logic behind it (Please note that numbers given here are not exact but should give you an idea.): If there is only one member (no IO), ServiceThread utilization will be about 95% and it will process between 30K and 120K operations per second, depending on the server. As the number of members in the cluster increases, IO Threads will be using more CPU and eventually ServiceThread will go down to 35% CPU utilization so ServiceThread will process between 10K and 40K operations per second. ServiceThread CPU utilization will be at about 35% regardless of the size of the cluster. (The only thing that can affect that would be the network utilization.) This also means that total number of operations processed by the entire cluster will be between N*10K and N*40K; N being the number of nodes in the cluster. About half of these operations will be backup operations (assuming one backup) so client threads will realize between N*5K and N*20K operations per second in total. Since there is only one thread accessing the data structures, increase in the number of nodes doesn't create any contention so access to the data structures will be always at the same speed. This is very important for Hazelcast's scalability.

This also makes writing code super easy because significant portion of the code is actually single-threaded so it is less error-prone and easily maintainable.

No synchronization or long running jobs are allowed in this thread. All operations running in this thread have to complete in microseconds.

3. InThread and OutThread:

Hazelcast separates reads and writes by having two separate threads; one for reading, and the other for writing. Eache IO thread uses its own NIO selector instance. InThread handles OP_ACCEPT and OP_READ socket operations while OutThread handles OP_CONNECT and OP_WRITE operations.

Each thread has its queue that they dequeue to register these operations with their selectors so operation registrations and operation processing happens in the same threads.

InThread's runnable is the com.hazelcast.nio.InSelector and OutThread's runnable is the com.hazelcast.nio.OutSelector. They both extends SelectorBase which constantly processes its registration queue ('selectorQueue') and the selectedKeys.

Members are connected to each other via TCP/IP. If there are N number of members in the cluster then there will be N*(N-1) connection end point and (N*(N-1))/2 connections. There can be only one connection between two members, meaning, if m2 creates a connection to m1, m1 doesn't create another connection to m2 and the rule here is that new members connect to the older members.

If you look at the com.hazelcast.nio.Connection, you will see that each connection is representing a socket channel and has com.hazelcast.nio.ReadHandler and WriteHandler instances which are attached to the socket channel's OP_READ and OP_WRITE operation selectionKeys respectively. When InSelector selects OP_READ selection key (when this operation is ready for the selector), InSelector will get the attached ReadHandler instance from the selectionKey and call its ReadHandler.handle() method. Same for the OutSelector. When OutSelector selects OP_WRITE selection key (when this operation is ready for the selector), OutSelector will get the attached WriteHandler instance from the selectionKey and call its WriteHandler.handle() method.

When ServiceThread wants to send an Invocation instance to a member, it will

see com.hazelcast.impl.BaseManager.send(Packet, Address).

see com.hazelcast.nio.SelectorBase.run().

Connections are always registered/interested for OP_READ operations. When InSelector is ready for reading from a socket channel, it will get the ReadHandler instance from the selectionKey and call its handle() method. handle() method will read Invocation instances from the underlying socket and when an Invocation instance is fully read, it will enqueue it into ServiceThread's (ClusterService class) queue to be processed.

4. MulticastThread:

If multicast discovery is enabled (this is the default), and node is the master (oldest member) in the cluster then MulticastThread is started to listen for join requests from the new members. When it receives join request (com.hazelcast.nio.MulticastService.JoinInfo class), it will check if the node is allowed to join, if so, it will send its address to the sender so that the sender node can create a TCP/IP connection to the master and send a JoinRequest.

5. Executor Threads:

Each node employs a local ExecutorService threads which handle the event listener calls and distributed executions. Number of core and max threads can be configured.

All your distributed objects such as your key and value objects, objects you offer into distributed queue and your distributed callable/runnable objects have to be Serializable.

Hazelcast serializes all your objects into an instance of com.hazelcast.nio.Data. Data is the binary representation of an object and it holds list of java.nio.ByteBuffer instances which are reused. When Hazelcast serializes an object into Data, it first checks whether the object is an instance of well-known, optimizable object such as String, Long, Integer, byte[], ByteBuffer, Date. If not, it then checks whether the object is an instance of com.hazelcast.nio.DataSerializable. If not, Hazelcast uses standard java serialization to convert the object into binary format. See com.hazelcast.nio.Serializer for details.

So for faster serialization, Hazelcast recommends to use of String, Long, Integer, byte[] objects or to implement com.hazelcast.nio.DataSerializable interface. Here is an example of a class implementing com.hazelcast.nio.DataSerializable interface.

public class Address implements com.hazelcast.nio.DataSerializable {
	private String street;
	private int zipCode;
	private String city;
	private String state;

	public Address() {}

	//getters setters..

	public void writeData(DataOutput out) throws IOException {
		out.writeUTF(street);
		out.writeInt(zipCode);
		out.writeUTF(city);
		out.writeUTF(state);
	}

	public void readData (DataInput in) throws IOException {
		street	= in.readUTF();
		zipCode = in.readInt();
		city	= in.readUTF();
		state	= in.readUTF();
	}
}

Lets take a look at another example which is encapsulating a DataSerializable field.

public class Employee implements com.hazelcast.nio.DataSerializable {
	private String firstName;
	private String lastName;
	private int age;
	private double salary;
	private Address address; //address itself is DataSerializable

	public Employee() {}

	//getters setters..

	public void writeData(DataOutput out) throws IOException {
		out.writeUTF(firstName);
		out.writeUTF(lastName);
		out.writeInt(age);
		out.writeDouble (salary);
		address.writeData (out);
	}

	public void readData (DataInput in) throws IOException {
		firstName = in.readUTF();
		lastName  = in.readUTF();
		age       = in.readInt();
		salary 	  = in.readDouble();
		address   = new Address();
		// since Address is DataSerializable let it read its own internal state
		address.readData (in);
	}
}

As you can see, since address field itself is DataSerializable, it is calling address.writeData(out) when writing and address.readData(in) when reading.

Caution: Hazelcast serialization is done on the user thread and it assumes that there will be only one object serialization at a time. So putting any Hazelcast operation that will require to serialize anything else will brake the serialization. For Example: Putting

Hazelcast.getMap("anyMap").put("key", "dummy value");

line in readData or writeData methods will breake the serialization. If you have to perform such an operation, at least it should be performed in another thread which will force the serialization to take on different thread.

It is important to note that Hazelcast is a peer to peer clustering so there is no 'master' kind of server in Hazelcast. Every member in the cluster is equal and has the same rights and responsibilities.

When a node starts up, it will check to see if there is already a cluster in the network. There are two ways to find this out:

If there is no existing node, then the node will be the first member of the cluster. If multicast is enabled then it will start a multicast listener so that it can respond to incoming join requests. Otherwise it will listen for join request coming via TCP/IP.

If there is an existing cluster already, then the oldest member in the cluster will receive the join request and check if the request is for the right group. If so, the oldest member in the cluster will start the join process.

In the join process, the oldest member will:

Every member in the cluster has the same member list in the same order. First member is the oldest member so if the oldest member dies, second member in the list becomes the first member in the list and the new oldest member.

See com.hazelcast.impl.Node and com.hazelcast.impl.ClusterManager for details.

Q. If, let say 50+, nodes are trying to join the cluster at the same time, are they going to join the cluster one by one?

No. As soon as the oldest member receives the first valid join request, it will wait 5 seconds for others to join so that it can join multiple members in one shot. If there is no new node willing to join for the next 5 seconds, then oldest member will start the join process. If a member leaves the cluster though, because of a JVM crash for example, cluster will immediately take action and oldest member will start the data recovery process.

Hazelcast distributed map is a peer to peer, partitioned implementation so entries put into the map will be almost evenly partitioned onto the existing members. Entries are partitioned according to their keys.

Every key is owned by a member. So every key-aware operation, such as put, remove, get is routed to the member owning the key.

Q. How does Hazelcast determine the owner of a key?

Hazelcast creates fixed number of virtual partitions (blocks). Partition count is set to 271 by default. Each key falls into one of these partitions. Each partition is owned/managed by a member. Oldest member of the cluster will assign the ownerships of the partitions and let every member know who owns which partitions. So at any given time, each member knows the owner member of a each partition. Hazelcast will convert your key object to com.hazelcast.nio.Data then calculate the partition of the owner: partition-of-the-key = hash(keyData) % PARTITION_COUNT. Since each member(JVM) knows the owner of each partition, each member can find out which member owns the key.

Q. Can I get the owner of a key?

Yes. Use Partition API to get the partition that your key falls into and then get the owner of that partition. Note that owner of the partition can change over time as new members join or existing members leave the cluster.

PartitionService partitionService = Hazelcast.getPartitionService();
Partition partition = partitionService.getPartition(key);
Member ownerMember = partition.getOwner();

Locally owned entries can be obtained by calling map.localKeySet().

Q. What happens when a new member joins?

Just like any other member in the cluster, the oldest member also knows who owns which partition and what the oldest member knows is always right. The oldest member is also responsible for redistributing the partition ownerships when a new member joins. Since there is new member, oldest member will take ownership of some of the partitions and give them to the new member. It will try to move the least amount of data possible. New ownership information of all partitions is then sent to all members.

Notice that the new ownership information may not reach each member at the same time and the cluster never stops responding to user map operations even during joins so if a member routes the operation to a wrong member, target member will tell the caller to re-do the operation.

If a member's partition is given to the new member, then the member will send all entries of that partition to the new member (Migrating the entries). Eventually every member in the cluster will own almost same number of partitions, and almost same number of entries. Also eventually every member will know the owner of each partition (and each key).

You can listen for migration events. MigrationEvent contains the partitionId, oldOwner, and newOwner information.

PartitionService partitionService = Hazelcast.getPartitionService();
partitionService.addMigrationListener(new MigrationListener () {

   public void migrationStarted(MigrationEvent migrationEvent) {
      System.out.println(migrationEvent);
   }

   public void migrationCompleted(MigrationEvent migrationEvent) {
      System.out.println(migrationEvent);
   }
});

Q. How about distributed set and list?

Both distributed set and list are implemented on top of distributed map. The underlying distributed map doesn't hold value; it only knows the key. Items added to both list and set are treated as keys. Unlike distributed set, since distributed list can have duplicate items, if an existing item is added again, copyCount of the entry (com.hazelcast.impl.ConcurrentMapManager.Record) is incremented. Also note that index based methods of distributed list, such as List.get(index) and List.indexOf(Object), are not supported because it is too costly to keep distributed indexes of list items so it is not worth implementing.

Check out the com.hazelcast.impl.ConcurrentMapManager class for the implementation. As you will see, the implementation is lock-free because ConcurrentMapManager is a singleton and processed by only one thread, the ServiceThread.

Hazelcast is the distributed implementation of several structures that exist in Java. Most of the time it behaves as you expect. However there are some design choices in Hazelcast that violate some contracts. This page will list those violations.

Hazelcast allows you to create more than one member on the same JVM. Each member is called HazelcastInstance and each will have its own configuration, socket and threads, so you can treat them as totally separate members. This enables us to write and run cluster unit tests on single JVM. As you can use this feature for creating separate members different applications running on the same JVM (imagine running multiple webapps on the same JVM), you can also use this feature for testing Hazelcast cluster.

Let's say you want to test if two members have the same size of a map.

   @Test
    public void testTwoMemberMapSizes() {
        // start the first member
        HazelcastInstance h1 = Hazelcast.newHazelcastInstance(null);
        // get the map and put 1000 entries
        Map map1 = h1.getMap("testmap");
        for (int i = 0; i < 1000; i++) {
            map1.put(i, "value" + i);
        }
        // check the map size
        assertEquals(1000, map1.size());
        // start the second member
        HazelcastInstance h2 = Hazelcast.newHazelcastInstance(null);
        // get the same map from the second member
        Map map2 = h2.getMap("testmap");
        // check the size of map2
        assertEquals(1000, map2.size());
        // check the size of map1 again
        assertEquals(1000, map1.size());
    }

In the test above, everything happened in the same thread. When developing multi-threaded test, coordination of the thread executions has to be carefully handled. Usage of CountDownLatch for thread coordination is highly recommended. You can certainly use other things. Here is an example where we need to listen for messages and make sure that we got these messages:

   @Test
    public void testTopic() {
        // start two member cluster
        HazelcastInstance h1 = Hazelcast.newHazelcastInstance(null);
        HazelcastInstance h2 = Hazelcast.newHazelcastInstance(null);
        String topicName = "TestMessages";
        // get a topic from the first member and add a messageListener
        ITopic<String> topic1 = h1.getTopic(topicName);
        final CountDownLatch latch1 = new CountDownLatch(1);
        topic1.addMessageListener(new MessageListener() {
            public void onMessage(Object msg) {
                assertEquals("Test1", msg);
                latch1.countDown();
            }
        });
        // get a topic from the second member and add a messageListener
        ITopic<String> topic2 = h2.getTopic(topicName);
        final CountDownLatch latch2 = new CountDownLatch(2);
        topic2.addMessageListener(new MessageListener() {
            public void onMessage(Object msg) {
                assertEquals("Test1", msg);
                latch2.countDown();
            }
        });
        // publish the first message, both should receive this
        topic1.publish("Test1");
        // shutdown the first member
        h1.shutdown();
        // publish the second message, second member's topic should receive this
        topic2.publish("Test1");
        try {
            // assert that the first member's topic got the message
            assertTrue(latch1.await(5, TimeUnit.SECONDS));
            // assert that the second members' topic got two messages
            assertTrue(latch2.await(5, TimeUnit.SECONDS));
        } catch (InterruptedException ignored) {
        }
    }

You can surely start Hazelcast members with different configuration. Let's say we want to test if Hazelcast SuperClient can shutdown fine.

   @Test(timeout = 60000)
    public void shutdownSuperClient() {
        // first config for normal cluster member
        Config c1 = new XmlConfigBuilder().build();
        c1.setPortAutoIncrement(false);
        c1.setPort(5709);
        // second config for super client
        Config c2 = new XmlConfigBuilder().build();
        c2.setPortAutoIncrement(false);
        c2.setPort(5710);
        // make sure to super client = true
        c2.setSuperClient(true);
        // start the normal member with c1
        HazelcastInstance hNormal = Hazelcast.newHazelcastInstance(c1);
        // start the super client with different configuration c2
        HazelcastInstance hSuper = Hazelcast.newHazelcastInstance(c2);
        hNormal.getMap("default").put("1", "first");
        assert hSuper.getMap("default").get("1").equals("first");
        hNormal.shutdown();
        hSuper.shutdown();
    }

Also remember to call Hazelcast.shutdownAll() after each test case to make sure that there is no other running member left from the previous tests.

   @After
    public void cleanup() throws Exception {
        Hazelcast.shutdownAll(); 
    }

Need more info? Check out existing tests.

Random order of planned features.

History of existing features is available atRelease Notes.

Please see, Todo page for planned features.

1.9.4

1.9.3

1.9

1.8.4

1.8.3

1.8.2

1.8.1

1.8

1.7.1

1.7

1.6

1.5

1.4

1.3

1.2

1.1

1.0