001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *      http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018
019package org.apache.oozie.service;
020
021import java.util.Collection;
022import java.util.Date;
023import java.util.HashSet;
024import java.util.Iterator;
025import java.util.Map;
026import java.util.Set;
027import java.util.concurrent.ConcurrentHashMap;
028import java.util.concurrent.ConcurrentMap;
029
030import org.apache.hadoop.conf.Configuration;
031import org.apache.hadoop.util.ReflectionUtils;
032import org.apache.oozie.CoordinatorActionBean;
033import org.apache.oozie.client.CoordinatorAction;
034import org.apache.oozie.command.coord.CoordActionUpdatePushMissingDependency;
035import org.apache.oozie.dependency.hcat.HCatDependencyCache;
036import org.apache.oozie.dependency.hcat.SimpleHCatDependencyCache;
037import org.apache.oozie.executor.jpa.CoordActionQueryExecutor;
038import org.apache.oozie.executor.jpa.JPAExecutorException;
039import org.apache.oozie.executor.jpa.CoordActionQueryExecutor.CoordActionQuery;
040import org.apache.oozie.util.HCatURI;
041import org.apache.oozie.util.XLog;
042
043import com.google.common.annotations.VisibleForTesting;
044
045/**
046 * Module that functions like a caching service to maintain partition dependency mappings
047 */
048public class PartitionDependencyManagerService implements Service {
049
050    public static final String CONF_PREFIX = Service.CONF_PREFIX + "PartitionDependencyManagerService.";
051    public static final String CACHE_MANAGER_IMPL = CONF_PREFIX + "cache.manager.impl";
052    public static final String CACHE_PURGE_INTERVAL = CONF_PREFIX + "cache.purge.interval";
053    public static final String CACHE_PURGE_TTL = CONF_PREFIX + "cache.purge.ttl";
054
055    private static XLog LOG = XLog.getLog(PartitionDependencyManagerService.class);
056
057    private HCatDependencyCache dependencyCache;
058
059    /**
060     * Keep timestamp when missing dependencies of a coord action are registered
061     */
062    private ConcurrentMap<String, Long> registeredCoordActionMap;
063
064    private boolean purgeEnabled = false;
065
066    @Override
067    public void init(Services services) throws ServiceException {
068        init(services.getConf());
069    }
070
071    private void init(Configuration conf) throws ServiceException {
072        Class<?> defaultClass = conf.getClass(CACHE_MANAGER_IMPL, null);
073        dependencyCache = (defaultClass == null) ? new SimpleHCatDependencyCache()
074                : (HCatDependencyCache) ReflectionUtils.newInstance(defaultClass, null);
075        dependencyCache.init(conf);
076        LOG.info("PartitionDependencyManagerService initialized. Dependency cache is {0} ", dependencyCache.getClass()
077                .getName());
078        purgeEnabled = Services.get().get(JobsConcurrencyService.class).isHighlyAvailableMode();
079        if (purgeEnabled) {
080            Runnable purgeThread = new CachePurgeWorker(dependencyCache);
081            // schedule runnable by default every 10 min
082            Services.get()
083                    .get(SchedulerService.class)
084                    .schedule(purgeThread, 10, Services.get().getConf().getInt(CACHE_PURGE_INTERVAL, 600),
085                            SchedulerService.Unit.SEC);
086            registeredCoordActionMap = new ConcurrentHashMap<String, Long>();
087        }
088    }
089
090    private class CachePurgeWorker implements Runnable {
091        HCatDependencyCache cache;
092        public CachePurgeWorker(HCatDependencyCache cache) {
093            this.cache = cache;
094        }
095
096        @Override
097        public void run() {
098            if (Thread.currentThread().isInterrupted()) {
099                return;
100            }
101            try {
102                purgeMissingDependency(Services.get().getConf().getInt(CACHE_PURGE_TTL, 1800));
103            }
104            catch (Throwable error) {
105                XLog.getLog(PartitionDependencyManagerService.class).debug("Throwable in CachePurgeWorker thread run : ", error);
106            }
107        }
108
109        private void purgeMissingDependency(int timeToLive) {
110            long currentTime = new Date().getTime();
111            Set<String> staleActions = new HashSet<String>();
112            Iterator<String> actionItr = registeredCoordActionMap.keySet().iterator();
113            while(actionItr.hasNext()){
114                String actionId = actionItr.next();
115                Long regTime = registeredCoordActionMap.get(actionId);
116                if(regTime < (currentTime - timeToLive * 1000)){
117                    CoordinatorActionBean caBean = null;
118                    try {
119                        caBean = CoordActionQueryExecutor.getInstance().get(CoordActionQuery.GET_COORD_ACTION_STATUS, actionId);
120                    }
121                    catch (JPAExecutorException e) {
122                        LOG.warn("Error in checking coord action:" + actionId + "to purge, skipping", e);
123                    }
124                    if(caBean != null && !caBean.getStatus().equals(CoordinatorAction.Status.WAITING)){
125                        staleActions.add(actionId);
126                        actionItr.remove();
127                    }
128                }
129            }
130            dependencyCache.removeNonWaitingCoordActions(staleActions);
131        }
132    }
133
134    @Override
135    public void destroy() {
136        dependencyCache.destroy();
137    }
138
139    @Override
140    public Class<? extends Service> getInterface() {
141        return PartitionDependencyManagerService.class;
142    }
143
144    /**
145     * Add a missing partition dependency and the actionID waiting on it
146     *
147     * @param hcatURI dependency URI
148     * @param actionID ID of action which is waiting for the dependency
149     */
150    public void addMissingDependency(HCatURI hcatURI, String actionID) {
151        if (purgeEnabled) {
152            registeredCoordActionMap.put(actionID, new Date().getTime());
153        }
154        dependencyCache.addMissingDependency(hcatURI, actionID);
155    }
156
157    /**
158     * Remove a missing partition dependency associated with a actionID
159     *
160     * @param hcatURI dependency URI
161     * @param actionID ID of action which is waiting for the dependency
162     * @return true if successful, else false
163     */
164    public boolean removeMissingDependency(HCatURI hcatURI, String actionID) {
165        return dependencyCache.removeMissingDependency(hcatURI, actionID);
166    }
167
168    /**
169     * Get the list of actionIDs waiting for a partition
170     *
171     * @param hcatURI dependency URI
172     * @return list of actionIDs
173     */
174    public Collection<String> getWaitingActions(HCatURI hcatURI) {
175        return dependencyCache.getWaitingActions(hcatURI);
176    }
177
178    /**
179     * Mark a partition dependency as available
180     *
181     * @param server host:port of the server
182     * @param db name of the database
183     * @param table name of the table
184     * @param partitions list of available partitions
185     * @return list of actionIDs for which the dependency is now available
186     */
187    public void partitionAvailable(String server, String db, String table, Map<String, String> partitions) {
188        Collection<String> actionsWithAvailableDep = dependencyCache.markDependencyAvailable(server, db, table,
189                partitions);
190        if (actionsWithAvailableDep != null) {
191            for (String actionID : actionsWithAvailableDep) {
192                boolean ret = Services.get().get(CallableQueueService.class)
193                        .queue(new CoordActionUpdatePushMissingDependency(actionID), 100);
194                if (ret == false) {
195                    XLog.getLog(getClass()).warn(
196                            "Unable to queue the callable commands for PartitionDependencyManagerService for actionID "
197                                    + actionID + ".Most possibly command queue is full. Queue size is :"
198                                    + Services.get().get(CallableQueueService.class).queueSize());
199                }
200            }
201        }
202    }
203
204    /**
205     * Get a list of available dependency URIs for a actionID
206     *
207     * @param actionID action id
208     * @return list of available dependency URIs
209     */
210    public Collection<String> getAvailableDependencyURIs(String actionID) {
211        return dependencyCache.getAvailableDependencyURIs(actionID);
212    }
213
214    /**
215     * Remove the list of available dependency URIs for a actionID once the missing dependencies are processed.
216     *
217     * @param actionID action id
218     * @param dependencyURIs set of dependency URIs
219     * @return true if successful, else false
220     */
221    public boolean removeAvailableDependencyURIs(String actionID, Collection<String> dependencyURIs) {
222        return dependencyCache.removeAvailableDependencyURIs(actionID, dependencyURIs);
223    }
224
225    /**
226     * Remove a coord action from dependency cache when all push missing dependencies available
227     *
228     * @param actionID action id
229     * @param dependencyURIs set of dependency URIs
230     * @return true if successful, else false
231     */
232    public void removeCoordActionWithDependenciesAvailable(String actionID) {
233        if (purgeEnabled) {
234            registeredCoordActionMap.remove(actionID);
235        }
236        dependencyCache.removeCoordActionWithDependenciesAvailable(actionID);
237    }
238
239    @VisibleForTesting
240    public void runCachePurgeWorker() {
241        new CachePurgeWorker(dependencyCache).run();
242    }
243}