Allowing component to reattached to previously created map in DatabaseManager

Change-Id: Ie576171392481df726e62365f70d0e934e0ad954

@@ -415,16 +415,19 @@ public class DatabaseManager implements StorageService, StorageAdminService {
         getTransactions().stream().forEach(transactionManager::execute);
     }
 
-    protected <K, V> void registerMap(DefaultAsyncConsistentMap<K, V> map) {
-        // TODO: Support multiple local instances of the same map.
-        if (maps.putIfAbsent(map.name(), map) != null) {
-            throw new IllegalStateException("Map by name " + map.name() + " already exists");
+    protected <K, V> DefaultAsyncConsistentMap<K, V> registerMap(DefaultAsyncConsistentMap<K, V> map) {
+        DefaultAsyncConsistentMap<K, V> existing = maps.putIfAbsent(map.name(), map);
+        if (existing != null) {
+            // FIXME: We need to cleanly support different map instances with same name.
+            log.info("Map by name {} already exists", map.name());
+            return existing;
         }
 
         clusterCommunicator.<MapEvent<K, V>>addSubscriber(mapUpdatesSubject(map.name()),
                 map.serializer()::decode,
                 map::notifyLocalListeners,
                 eventDispatcher);
+        return map;
     }
 
     protected <K, V> void unregisterMap(DefaultAsyncConsistentMap<K, V> map) {

@@ -77,7 +77,6 @@ public class DefaultConsistentMapBuilder<K, V> implements ConsistentMapBuilder<K
                 event -> manager.clusterCommunicator.<MapEvent<K, V>>broadcast(event,
                         DatabaseManager.mapUpdatesSubject(name),
                         serializer::encode));
-        manager.registerMap(asyncMap);
-        return asyncMap;
+        return manager.registerMap(asyncMap);
     }
 }
\ No newline at end of file