pulsar-broker/src/main/java/com/yahoo/pulsar/broker/namespace/OwnershipCache.java - pulsar - Git at Google

 /**
  * Copyright 2016 Yahoo Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package com.yahoo.pulsar.broker.namespace;

 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkState;

 import java.util.Map;

 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.KeeperException.NoNodeException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.cache.CacheBuilder;
 import com.google.common.cache.CacheLoader;
 import com.google.common.cache.LoadingCache;
 import com.google.common.cache.RemovalListener;
 import com.google.common.cache.RemovalNotification;
 import com.yahoo.pulsar.broker.PulsarService;
 import com.yahoo.pulsar.broker.ServiceConfiguration;
 import com.yahoo.pulsar.common.naming.NamespaceBundle;
 import com.yahoo.pulsar.common.naming.NamespaceBundleFactory;
 import com.yahoo.pulsar.common.naming.NamespaceBundles;
 import com.yahoo.pulsar.common.naming.ServiceUnitId;
 import com.yahoo.pulsar.common.util.ObjectMapperFactory;
 import com.yahoo.pulsar.zookeeper.ZooKeeperCache;
 import com.yahoo.pulsar.zookeeper.ZooKeeperDataCache;

 /**
  * This class provides a cache service for all the service unit ownership among the brokers. It provide a cache service
  * as well as ZooKeeper read/write functions for a) lookup of a service unit ownership to a broker; b) take ownership of
  * a service unit by the local broker
  *
  *
  */
 public class OwnershipCache {

     private static final Logger LOG = LoggerFactory.getLogger(OwnershipCache.class);

     /**
      * The local broker URL that this <code>OwnershipCache</code> will set as owner
      */
     private final String ownerBrokerUrl;

     /**
      * The local broker URL that this <code>OwnershipCache</code> will set as owner
      */
     private final String ownerBrokerUrlTls;

     /**
      * The NamespaceEphemeralData objects that can be associated with the current owner
      */
     private final NamespaceEphemeralData[] selfOwnerInfos;

     /**
      * Service unit ownership cache of <code>ZooKeeper</code> data of ephemeral nodes showing all known ownership of
      * service unit to active brokers
      */
     private final ZooKeeperDataCache<NamespaceEphemeralData> ownershipReadOnlyCache;

     /**
      * The loading cache of locally owned <code>ServiceUnit</code> objects
      */
     private final LoadingCache<String, OwnedServiceUnit> ownedServiceUnitsCache;

     /**
      * The <code>ObjectMapper</code> to deserialize/serialize JSON objects
      */
     private final ObjectMapper jsonMapper = ObjectMapperFactory.create();

     /**
      * The <code>ZooKeeperCache</code> connecting to the local ZooKeeper
      */
     private final ZooKeeperCache localZkCache;

     /**
      * The <code>NamespaceBundleFactory</code> to construct <code>NamespaceBundles</code>
      */
     private final NamespaceBundleFactory bundleFactory;

     /**
      * The max number of retries to acquire the ownership in <code>ZooKeeper</code>
      */
     private static final int MAX_RETRIES_CREATE_ZNODE = 5; // tentatively decide to retry 5 times in acquiring the
                                                            // zookeeper node

     private class OwnedServiceUnitCacheLoader extends CacheLoader<String, OwnedServiceUnit> {
         @Override
         public OwnedServiceUnit load(String key) throws Exception {
             LOG.info("Acquiring zk lock on namespace {}", key);

             // Under the cache sync lock, acquiring the ZNode
             // If succeeded, we guaranteed that the cache entry is setup w/ ZNode acquired
             // Only enters this load function if ownedServiceUnitsCache does not have the key
             // invalidate the read-only cache to ensure loading the up-to-date znode
             int numTries = 0;
             while (numTries != MAX_RETRIES_CREATE_ZNODE) {
                 try {
                     checkArgument(ServiceUnitZkUtils
                             .acquireNameSpace(localZkCache.getZooKeeper(), key, selfOwnerInfos[0]).getNativeUrl()
                             .equals(ownerBrokerUrl)
                             || ServiceUnitZkUtils.acquireNameSpace(localZkCache.getZooKeeper(), key, selfOwnerInfos[0])
                                     .getNativeUrlTls().equals(ownerBrokerUrlTls));
                     ownershipReadOnlyCache.invalidate(key);
                     LOG.info("Acquired zk lock on namespace {}", key);
                     return new OwnedServiceUnit(ServiceUnitZkUtils.suIdFromPath(key, bundleFactory));
                 } catch (Exception e) {
                     // Failed to acquire the namespace, try to load the read-only cache since some other broker may have
                     // won the race
                     LOG.warn(String.format("Failed in acquiring the namespace ownership. key=%s", key));
                     try {
                         ownershipReadOnlyCache.invalidate(key);
                         ownershipReadOnlyCache.get(key);
                         // if successful, the znode has been created by someone. break out the loop
                         break;
                     } catch (NoNodeException nne) {
                         // load read-only cache failed due to no node exists, hence, we should try to acquire the
                         // namespace
                         // again
                         numTries++;
                     } catch (Exception moreErr) {
                         // Other unexpected failure
                         LOG.error(String.format(
                                 "Unexpected exception while loading the read-only ZK cache for namespace. key=%s", key),
                                 moreErr);
                         throw moreErr;
                     }
                 }
             }
             checkArgument(numTries < MAX_RETRIES_CREATE_ZNODE,
                     "Maximum retries exceeded to acquire the namespace. key=" + key);
             // load the read-only cache w/ update-to-date entry
             checkArgument(ownershipReadOnlyCache.get(key).getNativeUrl().equals(ownerBrokerUrl),
                     "Some other broker acquired the namespace. key=" + key);
             return new OwnedServiceUnit(ServiceUnitZkUtils.suIdFromPath(key, bundleFactory));
         }
     }

     /**
      * Constructor of <code>OwnershipCache</code>
      *
      * @param ownerUrl
      *            the local broker URL that will be set as owner for the <code>ServiceUnit</code>
      */
     public OwnershipCache(PulsarService pulsar, NamespaceBundleFactory bundleFactory) {
         ServiceConfiguration conf = pulsar.getConfiguration();
         this.ownerBrokerUrl = pulsar.getBrokerServiceUrl();
         this.ownerBrokerUrlTls = pulsar.getBrokerServiceUrlTls();
         this.selfOwnerInfos = new NamespaceEphemeralData[] {
                 new NamespaceEphemeralData(ownerBrokerUrl, ownerBrokerUrlTls, pulsar.getWebServiceAddress(),
                         pulsar.getWebServiceAddressTls(), false),
                 new NamespaceEphemeralData(ownerBrokerUrl, ownerBrokerUrlTls, pulsar.getWebServiceAddress(),
                         pulsar.getWebServiceAddressTls(), true) };
         this.bundleFactory = bundleFactory;
         this.localZkCache = pulsar.getLocalZkCache();
         this.ownershipReadOnlyCache = pulsar.getLocalZkCacheService().ownerInfoCache();
         // ownedServiceUnitsCache contains all namespaces that are owned by the local broker
         this.ownedServiceUnitsCache = CacheBuilder.newBuilder().build(new OwnedServiceUnitCacheLoader());
     }

     /**
      * Method to get the current owner of the <code>ServiceUnit</code>
      *
      * @param suId
      *            identifier of the <code>ServiceUnit</code>
      * @return The ephemeral node data showing the current ownership info in <code>ZooKeeper</code>
      * @throws Exception
      *             throws exception if no ownership info is found
      */
     public NamespaceEphemeralData getOwner(ServiceUnitId suname) throws Exception {
         return this.ownershipReadOnlyCache.get(ServiceUnitZkUtils.path(suname));
     }

     /**
      * Method to get the current owner of the <code>ServiceUnit</code> or set the local broker as the owner if absent
      *
      * @param suId
      *            identifier of the <code>ServiceUnit</code>
      * @return The ephemeral node data showing the current ownership info in <code>ZooKeeper</code>
      * @throws Exception
      */
     public NamespaceEphemeralData getOrSetOwner(ServiceUnitId suname) throws Exception {
         String path = ServiceUnitZkUtils.path(suname);

         // If the node has been deleted between two checks, we need to try again
         while (true) {
             try {
                 // Trying to load the ownedServiceUnitCache by acquiring the ZNode via the loader
                 this.ownedServiceUnitsCache.get(path);
             } catch (Exception e) {
                 LOG.info(String.format("Failed in acquiring the ownership of service unit %s", suname), e);
             }

             try {
                 return this.ownershipReadOnlyCache.get(path);
             } catch (KeeperException.NoNodeException e) {
                 LOG.warn("Failed in getting service unit from read-only cache {}", suname);
             }
         }
     }

     /**
      * Method to remove the ownership of local broker on the <code>ServiceUnit</code>, if owned
      *
      * @param suId
      *            identifier of the <code>ServiceUnit</code>
      * @throws Exception
      */
     public void removeOwnership(ServiceUnitId suname) throws Exception {
         // Under the cache sync lock, removing the ZNode
         // If succeeded, we guaranteed that the cache entry is removed together w/ ZNode
         String key = ServiceUnitZkUtils.path(suname);
         try {
             localZkCache.getZooKeeper().delete(key, -1);
             this.ownedServiceUnitsCache.invalidate(key);
             this.ownershipReadOnlyCache.invalidate(key);
             LOG.info("Removed zk lock for service unit: {}", key);
         } catch (KeeperException.NoNodeException e) {
             LOG.warn("[{}] Failed to delete the namespace ephemeral node. key={}", key, e.getMessage());
         }

     }

     /**
      * Method to remove ownership of all owned bundles
      *
      * @param bundles
      *            <code>NamespaceBundles</code> to remove from ownership cache
      */
     public void removeOwnership(NamespaceBundles bundles) {
         boolean hasError = false;
         for (NamespaceBundle bundle : bundles.getBundles()) {
             if (getOwnedServiceUnit(bundle) == null) {
                 // continue
                 continue;
             }
             try {
                 this.removeOwnership(bundle);
             } catch (Exception e) {
                 LOG.warn(String.format("Failed to remove ownership of a service unit: %s", bundle), e);
                 hasError = true;
             }
         }
         checkState(!hasError, "Not able to remove all owned bundles");
     }

     /**
      * Method to access the map of all <code>ServiceUnit</code> objects owned by the local broker
      *
      * @return a map of owned <code>ServiceUnit</code> objects
      */
     public Map<String, OwnedServiceUnit> getOwnedServiceUnits() {
         return this.ownedServiceUnitsCache.asMap();
     }

     public OwnedServiceUnit getOwnedServiceUnit(ServiceUnitId suname) {
         return this.ownedServiceUnitsCache.getIfPresent(ServiceUnitZkUtils.path(suname));
     }

     public void disableOwnership(ServiceUnitId suName) throws Exception {
         String path = ServiceUnitZkUtils.path(suName);
         localZkCache.getZooKeeper().setData(path, jsonMapper.writeValueAsBytes(selfOwnerInfos[1]), -1);
         this.ownershipReadOnlyCache.invalidate(path);
     }

     /**
      * Update bundle state in a local cache
      *
      * @param bundle
      * @throws Exception
      */
     public void updateBundleState(ServiceUnitId bundle, boolean isActive) throws Exception {
         // Disable owned instance in local cache
         OwnedServiceUnit ownedServiceUnit = getOwnedServiceUnit(bundle);
         if (ownedServiceUnit != null) {
             ownedServiceUnit.setActive(isActive);
         }
     }
 }
	/**
	* Copyright 2016 Yahoo Inc.
	*
	* Licensed under the Apache License, Version 2.0 (the "License");
	* you may not use this file except in compliance with the License.
	* You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package com.yahoo.pulsar.broker.namespace;

	import static com.google.common.base.Preconditions.checkArgument;
	import static com.google.common.base.Preconditions.checkState;

	import java.util.Map;

	import org.apache.zookeeper.KeeperException;
	import org.apache.zookeeper.KeeperException.NoNodeException;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import com.fasterxml.jackson.databind.ObjectMapper;
	import com.google.common.cache.CacheBuilder;
	import com.google.common.cache.CacheLoader;
	import com.google.common.cache.LoadingCache;
	import com.google.common.cache.RemovalListener;
	import com.google.common.cache.RemovalNotification;
	import com.yahoo.pulsar.broker.PulsarService;
	import com.yahoo.pulsar.broker.ServiceConfiguration;
	import com.yahoo.pulsar.common.naming.NamespaceBundle;
	import com.yahoo.pulsar.common.naming.NamespaceBundleFactory;
	import com.yahoo.pulsar.common.naming.NamespaceBundles;
	import com.yahoo.pulsar.common.naming.ServiceUnitId;
	import com.yahoo.pulsar.common.util.ObjectMapperFactory;
	import com.yahoo.pulsar.zookeeper.ZooKeeperCache;
	import com.yahoo.pulsar.zookeeper.ZooKeeperDataCache;

	/**
	* This class provides a cache service for all the service unit ownership among the brokers. It provide a cache service
	* as well as ZooKeeper read/write functions for a) lookup of a service unit ownership to a broker; b) take ownership of
	* a service unit by the local broker
	*
	*
	*/
	public class OwnershipCache {

	private static final Logger LOG = LoggerFactory.getLogger(OwnershipCache.class);

	/**
	* The local broker URL that this <code>OwnershipCache</code> will set as owner
	*/
	private final String ownerBrokerUrl;

	/**
	* The local broker URL that this <code>OwnershipCache</code> will set as owner
	*/
	private final String ownerBrokerUrlTls;

	/**
	* The NamespaceEphemeralData objects that can be associated with the current owner
	*/
	private final NamespaceEphemeralData[] selfOwnerInfos;

	/**
	* Service unit ownership cache of <code>ZooKeeper</code> data of ephemeral nodes showing all known ownership of
	* service unit to active brokers
	*/
	private final ZooKeeperDataCache<NamespaceEphemeralData> ownershipReadOnlyCache;

	/**
	* The loading cache of locally owned <code>ServiceUnit</code> objects
	*/
	private final LoadingCache<String, OwnedServiceUnit> ownedServiceUnitsCache;

	/**
	* The <code>ObjectMapper</code> to deserialize/serialize JSON objects
	*/
	private final ObjectMapper jsonMapper = ObjectMapperFactory.create();

	/**
	* The <code>ZooKeeperCache</code> connecting to the local ZooKeeper
	*/
	private final ZooKeeperCache localZkCache;

	/**
	* The <code>NamespaceBundleFactory</code> to construct <code>NamespaceBundles</code>
	*/
	private final NamespaceBundleFactory bundleFactory;

	/**
	* The max number of retries to acquire the ownership in <code>ZooKeeper</code>
	*/
	private static final int MAX_RETRIES_CREATE_ZNODE = 5; // tentatively decide to retry 5 times in acquiring the
	// zookeeper node

	private class OwnedServiceUnitCacheLoader extends CacheLoader<String, OwnedServiceUnit> {
	@Override
	public OwnedServiceUnit load(String key) throws Exception {
	LOG.info("Acquiring zk lock on namespace {}", key);

	// Under the cache sync lock, acquiring the ZNode
	// If succeeded, we guaranteed that the cache entry is setup w/ ZNode acquired
	// Only enters this load function if ownedServiceUnitsCache does not have the key
	// invalidate the read-only cache to ensure loading the up-to-date znode
	int numTries = 0;
	while (numTries != MAX_RETRIES_CREATE_ZNODE) {
	try {
	checkArgument(ServiceUnitZkUtils
	.acquireNameSpace(localZkCache.getZooKeeper(), key, selfOwnerInfos[0]).getNativeUrl()
	.equals(ownerBrokerUrl)
	\|\| ServiceUnitZkUtils.acquireNameSpace(localZkCache.getZooKeeper(), key, selfOwnerInfos[0])
	.getNativeUrlTls().equals(ownerBrokerUrlTls));
	ownershipReadOnlyCache.invalidate(key);
	LOG.info("Acquired zk lock on namespace {}", key);
	return new OwnedServiceUnit(ServiceUnitZkUtils.suIdFromPath(key, bundleFactory));
	} catch (Exception e) {
	// Failed to acquire the namespace, try to load the read-only cache since some other broker may have
	// won the race
	LOG.warn(String.format("Failed in acquiring the namespace ownership. key=%s", key));
	try {
	ownershipReadOnlyCache.invalidate(key);
	ownershipReadOnlyCache.get(key);
	// if successful, the znode has been created by someone. break out the loop
	break;
	} catch (NoNodeException nne) {
	// load read-only cache failed due to no node exists, hence, we should try to acquire the
	// namespace
	// again
	numTries++;
	} catch (Exception moreErr) {
	// Other unexpected failure
	LOG.error(String.format(
	"Unexpected exception while loading the read-only ZK cache for namespace. key=%s", key),
	moreErr);
	throw moreErr;
	}
	}
	}
	checkArgument(numTries < MAX_RETRIES_CREATE_ZNODE,
	"Maximum retries exceeded to acquire the namespace. key=" + key);
	// load the read-only cache w/ update-to-date entry
	checkArgument(ownershipReadOnlyCache.get(key).getNativeUrl().equals(ownerBrokerUrl),
	"Some other broker acquired the namespace. key=" + key);
	return new OwnedServiceUnit(ServiceUnitZkUtils.suIdFromPath(key, bundleFactory));
	}
	}

	/**
	* Constructor of <code>OwnershipCache</code>
	*
	* @param ownerUrl
	* the local broker URL that will be set as owner for the <code>ServiceUnit</code>
	*/
	public OwnershipCache(PulsarService pulsar, NamespaceBundleFactory bundleFactory) {
	ServiceConfiguration conf = pulsar.getConfiguration();
	this.ownerBrokerUrl = pulsar.getBrokerServiceUrl();
	this.ownerBrokerUrlTls = pulsar.getBrokerServiceUrlTls();
	this.selfOwnerInfos = new NamespaceEphemeralData[] {
	new NamespaceEphemeralData(ownerBrokerUrl, ownerBrokerUrlTls, pulsar.getWebServiceAddress(),
	pulsar.getWebServiceAddressTls(), false),
	new NamespaceEphemeralData(ownerBrokerUrl, ownerBrokerUrlTls, pulsar.getWebServiceAddress(),
	pulsar.getWebServiceAddressTls(), true) };
	this.bundleFactory = bundleFactory;
	this.localZkCache = pulsar.getLocalZkCache();
	this.ownershipReadOnlyCache = pulsar.getLocalZkCacheService().ownerInfoCache();
	// ownedServiceUnitsCache contains all namespaces that are owned by the local broker
	this.ownedServiceUnitsCache = CacheBuilder.newBuilder().build(new OwnedServiceUnitCacheLoader());
	}

	/**
	* Method to get the current owner of the <code>ServiceUnit</code>
	*
	* @param suId
	* identifier of the <code>ServiceUnit</code>
	* @return The ephemeral node data showing the current ownership info in <code>ZooKeeper</code>
	* @throws Exception
	* throws exception if no ownership info is found
	*/
	public NamespaceEphemeralData getOwner(ServiceUnitId suname) throws Exception {
	return this.ownershipReadOnlyCache.get(ServiceUnitZkUtils.path(suname));
	}

	/**
	* Method to get the current owner of the <code>ServiceUnit</code> or set the local broker as the owner if absent
	*
	* @param suId
	* identifier of the <code>ServiceUnit</code>
	* @return The ephemeral node data showing the current ownership info in <code>ZooKeeper</code>
	* @throws Exception
	*/
	public NamespaceEphemeralData getOrSetOwner(ServiceUnitId suname) throws Exception {
	String path = ServiceUnitZkUtils.path(suname);

	// If the node has been deleted between two checks, we need to try again
	while (true) {
	try {
	// Trying to load the ownedServiceUnitCache by acquiring the ZNode via the loader
	this.ownedServiceUnitsCache.get(path);
	} catch (Exception e) {
	LOG.info(String.format("Failed in acquiring the ownership of service unit %s", suname), e);
	}

	try {
	return this.ownershipReadOnlyCache.get(path);
	} catch (KeeperException.NoNodeException e) {
	LOG.warn("Failed in getting service unit from read-only cache {}", suname);
	}
	}
	}

	/**
	* Method to remove the ownership of local broker on the <code>ServiceUnit</code>, if owned
	*
	* @param suId
	* identifier of the <code>ServiceUnit</code>
	* @throws Exception
	*/
	public void removeOwnership(ServiceUnitId suname) throws Exception {
	// Under the cache sync lock, removing the ZNode
	// If succeeded, we guaranteed that the cache entry is removed together w/ ZNode
	String key = ServiceUnitZkUtils.path(suname);
	try {
	localZkCache.getZooKeeper().delete(key, -1);
	this.ownedServiceUnitsCache.invalidate(key);
	this.ownershipReadOnlyCache.invalidate(key);
	LOG.info("Removed zk lock for service unit: {}", key);
	} catch (KeeperException.NoNodeException e) {
	LOG.warn("[{}] Failed to delete the namespace ephemeral node. key={}", key, e.getMessage());
	}

	}

	/**
	* Method to remove ownership of all owned bundles
	*
	* @param bundles
	* <code>NamespaceBundles</code> to remove from ownership cache
	*/
	public void removeOwnership(NamespaceBundles bundles) {
	boolean hasError = false;
	for (NamespaceBundle bundle : bundles.getBundles()) {
	if (getOwnedServiceUnit(bundle) == null) {
	// continue
	continue;
	}
	try {
	this.removeOwnership(bundle);
	} catch (Exception e) {
	LOG.warn(String.format("Failed to remove ownership of a service unit: %s", bundle), e);
	hasError = true;
	}
	}
	checkState(!hasError, "Not able to remove all owned bundles");
	}

	/**
	* Method to access the map of all <code>ServiceUnit</code> objects owned by the local broker
	*
	* @return a map of owned <code>ServiceUnit</code> objects
	*/
	public Map<String, OwnedServiceUnit> getOwnedServiceUnits() {
	return this.ownedServiceUnitsCache.asMap();
	}

	public OwnedServiceUnit getOwnedServiceUnit(ServiceUnitId suname) {
	return this.ownedServiceUnitsCache.getIfPresent(ServiceUnitZkUtils.path(suname));
	}

	public void disableOwnership(ServiceUnitId suName) throws Exception {
	String path = ServiceUnitZkUtils.path(suName);
	localZkCache.getZooKeeper().setData(path, jsonMapper.writeValueAsBytes(selfOwnerInfos[1]), -1);
	this.ownershipReadOnlyCache.invalidate(path);
	}

	/**
	* Update bundle state in a local cache
	*
	* @param bundle
	* @throws Exception
	*/
	public void updateBundleState(ServiceUnitId bundle, boolean isActive) throws Exception {
	// Disable owned instance in local cache
	OwnedServiceUnit ownedServiceUnit = getOwnedServiceUnit(bundle);
	if (ownedServiceUnit != null) {
	ownedServiceUnit.setActive(isActive);
	}
	}
	}