OwnershipCache implements TimerTask { public boolean updateOwner(String stream, SocketAddress addr) { SocketAddress oldAddr = stream2Addresses.putIfAbsent(stream, addr); if (null != oldAddr && oldAddr.equals(addr)) { return true; } if (null != oldAddr) { if (stream2Addresses.replace(stream, oldAddr, addr)) { logger.info("Storing ownership for stream : {}, old host : {}, new host : {}.", new Object[] { stream, oldAddr, addr }); StringBuilder sb = new StringBuilder(); sb.append("Ownership changed '") .append(oldAddr).append("' -> '").append(addr).append("'"); removeOwnerFromStream(stream, oldAddr, sb.toString()); ownershipStatsLogger.onRemove(stream); ownershipStatsLogger.onAdd(stream); } else { logger.warn("Ownership of stream : {} has been changed from {} to {} when storing host : {}.", new Object[] { stream, oldAddr, stream2Addresses.get(stream), addr }); return false; } } else { logger.info("Storing ownership for stream : {}, host : {}.", stream, addr); ownershipStatsLogger.onAdd(stream); } Set<String> streamsForHost = address2Streams.get(addr); if (null == streamsForHost) { Set<String> newStreamsForHost = new HashSet<String>(); streamsForHost = address2Streams.putIfAbsent(addr, newStreamsForHost); if (null == streamsForHost) { streamsForHost = newStreamsForHost; } } synchronized (streamsForHost) { if (addr.equals(stream2Addresses.get(stream))) { streamsForHost.add(stream); } } return true; } OwnershipCache(ClientConfig clientConfig,
                          HashedWheelTimer timer,
                          StatsReceiver statsReceiver,
                          StatsReceiver streamStatsReceiver); @Override void run(Timeout timeout); OwnershipStatsLogger getOwnershipStatsLogger(); boolean updateOwner(String stream, SocketAddress addr); SocketAddress getOwner(String stream); void removeOwnerFromStream(String stream, SocketAddress addr, String reason); void removeAllStreamsFromOwner(SocketAddress addr); int getNumCachedStreams(); Map<SocketAddress, Set<String>> getStreamOwnershipDistribution(); Map<String, SocketAddress> getStreamOwnerMapping();  }
@Test(timeout = 60000) public void testUpdateOwner() { OwnershipCache cache = createOwnershipCache(); SocketAddress addr = createSocketAddress(1000); String stream = runtime.getMethodName(); assertTrue("Should successfully update owner if no owner exists before", cache.updateOwner(stream, addr)); assertEquals("Owner should be " + addr + " for stream " + stream, addr, cache.getOwner(stream)); assertTrue("Should successfully update owner if old owner is same", cache.updateOwner(stream, addr)); assertEquals("Owner should be " + addr + " for stream " + stream, addr, cache.getOwner(stream)); }