Source code

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.executor.jpa;
020
021import java.sql.Timestamp;
022import java.util.ArrayList;
023import java.util.Date;
024import java.util.List;
025
026import javax.persistence.EntityManager;
027import javax.persistence.Query;
028
029import org.apache.oozie.CoordinatorActionBean;
030import org.apache.oozie.CoordinatorJobBean;
031import org.apache.oozie.ErrorCode;
032import org.apache.oozie.StringBlob;
033import org.apache.oozie.service.JPAService;
034import org.apache.oozie.service.Services;
035import org.apache.oozie.util.DateUtils;
036
037/**
038 * Query Executor that provides API to run query for Coordinator Job
039 */
040
041public class CoordJobQueryExecutor extends QueryExecutor<CoordinatorJobBean, CoordJobQueryExecutor.CoordJobQuery> {
042
043    public enum CoordJobQuery {
044        UPDATE_COORD_JOB,
045        UPDATE_COORD_JOB_STATUS,
046        UPDATE_COORD_JOB_BUNDLEID,
047        UPDATE_COORD_JOB_APPNAMESPACE,
048        UPDATE_COORD_JOB_STATUS_PENDING,
049        UPDATE_COORD_JOB_BUNDLEID_APPNAMESPACE_PAUSETIME,
050        UPDATE_COORD_JOB_STATUS_MODTIME,
051        UPDATE_COORD_JOB_STATUS_PENDING_MODTIME,
052        UPDATE_COORD_JOB_LAST_MODIFIED_TIME,
053        UPDATE_COORD_JOB_STATUS_PENDING_TIME,
054        UPDATE_COORD_JOB_MATERIALIZE,
055        UPDATE_COORD_JOB_CHANGE,
056        GET_COORD_JOB,
057        GET_COORD_JOB_USER_APPNAME,
058        GET_COORD_JOB_INPUT_CHECK,
059        GET_COORD_JOB_ACTION_READY,
060        GET_COORD_JOB_ACTION_KILL,
061        GET_COORD_JOB_MATERIALIZE,
062        GET_COORD_JOB_SUSPEND_KILL,
063        GET_COORD_JOB_STATUS,
064        GET_COORD_JOB_STATUS_PARENTID,
065        GET_COORD_JOBS_CHANGED,
066        GET_COORD_JOBS_OLDER_FOR_MATERILZATION,
067        GET_COORD_FOR_ABANDONEDCHECK,
068        GET_COORD_IDS_FOR_STATUS_TRANSIT
069    };
070
071    private static CoordJobQueryExecutor instance = new CoordJobQueryExecutor();
072
073    private CoordJobQueryExecutor() {
074    }
075
076    public static CoordJobQueryExecutor getInstance() {
077        return CoordJobQueryExecutor.instance;
078    }
079
080    @Override
081    public Query getUpdateQuery(CoordJobQuery namedQuery, CoordinatorJobBean cjBean, EntityManager em)
082            throws JPAExecutorException {
083        Query query = em.createNamedQuery(namedQuery.name());
084        switch (namedQuery) {
085            case UPDATE_COORD_JOB:
086                query.setParameter("appName", cjBean.getAppName());
087                query.setParameter("appPath", cjBean.getAppPath());
088                query.setParameter("concurrency", cjBean.getConcurrency());
089                query.setParameter("conf", cjBean.getConfBlob());
090                query.setParameter("externalId", cjBean.getExternalId());
091                query.setParameter("frequency", cjBean.getFrequency());
092                query.setParameter("lastActionNumber", cjBean.getLastActionNumber());
093                query.setParameter("timeOut", cjBean.getTimeout());
094                query.setParameter("timeZone", cjBean.getTimeZone());
095                query.setParameter("createdTime", cjBean.getCreatedTimestamp());
096                query.setParameter("endTime", cjBean.getEndTimestamp());
097                query.setParameter("execution", cjBean.getExecution());
098                query.setParameter("jobXml", cjBean.getJobXmlBlob());
099                query.setParameter("lastAction", cjBean.getLastActionTimestamp());
100                query.setParameter("lastModifiedTime", cjBean.getLastModifiedTimestamp());
101                query.setParameter("nextMaterializedTime", cjBean.getNextMaterializedTimestamp());
102                query.setParameter("origJobXml", cjBean.getOrigJobXmlBlob());
103                query.setParameter("slaXml", cjBean.getSlaXmlBlob());
104                query.setParameter("startTime", cjBean.getStartTimestamp());
105                query.setParameter("status", cjBean.getStatus().toString());
106                query.setParameter("timeUnit", cjBean.getTimeUnitStr());
107                query.setParameter("appNamespace", cjBean.getAppNamespace());
108                query.setParameter("bundleId", cjBean.getBundleId());
109                query.setParameter("matThrottling", cjBean.getMatThrottling());
110                query.setParameter("id", cjBean.getId());
111                break;
112            case UPDATE_COORD_JOB_STATUS:
113                query.setParameter("status", cjBean.getStatus().toString());
114                query.setParameter("id", cjBean.getId());
115                break;
116            case UPDATE_COORD_JOB_BUNDLEID:
117                query.setParameter("bundleId", cjBean.getBundleId());
118                query.setParameter("id", cjBean.getId());
119                break;
120            case UPDATE_COORD_JOB_APPNAMESPACE:
121                query.setParameter("appNamespace", cjBean.getAppNamespace());
122                query.setParameter("id", cjBean.getId());
123                break;
124            case UPDATE_COORD_JOB_STATUS_PENDING:
125                query.setParameter("status", cjBean.getStatus().toString());
126                query.setParameter("pending", cjBean.isPending() ? 1 : 0);
127                query.setParameter("id", cjBean.getId());
128                break;
129            case UPDATE_COORD_JOB_BUNDLEID_APPNAMESPACE_PAUSETIME:
130                query.setParameter("bundleId", cjBean.getBundleId());
131                query.setParameter("appNamespace", cjBean.getAppNamespace());
132                query.setParameter("pauseTime", cjBean.getPauseTimestamp());
133                query.setParameter("id", cjBean.getId());
134                break;
135            case UPDATE_COORD_JOB_STATUS_MODTIME:
136                query.setParameter("status", cjBean.getStatus().toString());
137                query.setParameter("lastModifiedTime", cjBean.getLastModifiedTimestamp());
138                query.setParameter("id", cjBean.getId());
139                break;
140            case UPDATE_COORD_JOB_STATUS_PENDING_MODTIME:
141                query.setParameter("status", cjBean.getStatus().toString());
142                query.setParameter("pending", cjBean.isPending() ? 1 : 0);
143                query.setParameter("lastModifiedTime", cjBean.getLastModifiedTimestamp());
144                query.setParameter("id", cjBean.getId());
145                break;
146            case UPDATE_COORD_JOB_LAST_MODIFIED_TIME:
147                query.setParameter("lastModifiedTime", cjBean.getLastModifiedTimestamp());
148                query.setParameter("id", cjBean.getId());
149                break;
150            case UPDATE_COORD_JOB_STATUS_PENDING_TIME:
151                query.setParameter("status", cjBean.getStatus().toString());
152                query.setParameter("pending", cjBean.isPending() ? 1 : 0);
153                query.setParameter("doneMaterialization", cjBean.isDoneMaterialization() ? 1 : 0);
154                query.setParameter("lastModifiedTime", cjBean.getLastModifiedTimestamp());
155                query.setParameter("suspendedTime", cjBean.getSuspendedTimestamp());
156                query.setParameter("id", cjBean.getId());
157                break;
158            case UPDATE_COORD_JOB_MATERIALIZE:
159                query.setParameter("status", cjBean.getStatus().toString());
160                query.setParameter("pending", cjBean.isPending() ? 1 : 0);
161                query.setParameter("doneMaterialization", cjBean.isDoneMaterialization() ? 1 : 0);
162                query.setParameter("lastActionTime", cjBean.getLastActionTimestamp());
163                query.setParameter("lastActionNumber", cjBean.getLastActionNumber());
164                query.setParameter("nextMatdTime", cjBean.getNextMaterializedTimestamp());
165                query.setParameter("id", cjBean.getId());
166                break;
167            case UPDATE_COORD_JOB_CHANGE:
168                query.setParameter("endTime", cjBean.getEndTimestamp());
169                query.setParameter("status", cjBean.getStatus().toString());
170                query.setParameter("pending", cjBean.isPending() ? 1 : 0);
171                query.setParameter("doneMaterialization", cjBean.isDoneMaterialization() ? 1 : 0);
172                query.setParameter("concurrency", cjBean.getConcurrency());
173                query.setParameter("pauseTime", cjBean.getPauseTimestamp());
174                query.setParameter("lastActionNumber", cjBean.getLastActionNumber());
175                query.setParameter("lastActionTime", cjBean.getLastActionTimestamp());
176                query.setParameter("nextMatdTime", cjBean.getNextMaterializedTimestamp());
177                query.setParameter("lastModifiedTime", cjBean.getLastModifiedTimestamp());
178                query.setParameter("id", cjBean.getId());
179                break;
180            default:
181                throw new JPAExecutorException(ErrorCode.E0603, "QueryExecutor cannot set parameters for "
182                        + namedQuery.name());
183        }
184        return query;
185    }
186
187    @Override
188    public Query getSelectQuery(CoordJobQuery namedQuery, EntityManager em, Object... parameters)
189            throws JPAExecutorException {
190        Query query = em.createNamedQuery(namedQuery.name());
191        switch (namedQuery) {
192            case GET_COORD_JOB:
193            case GET_COORD_JOB_USER_APPNAME:
194            case GET_COORD_JOB_INPUT_CHECK:
195            case GET_COORD_JOB_ACTION_READY:
196            case GET_COORD_JOB_ACTION_KILL:
197            case GET_COORD_JOB_MATERIALIZE:
198            case GET_COORD_JOB_SUSPEND_KILL:
199            case GET_COORD_JOB_STATUS:
200            case GET_COORD_JOB_STATUS_PARENTID:
201                query.setParameter("id", parameters[0]);
202                break;
203            case GET_COORD_JOBS_CHANGED:
204                query.setParameter("lastModifiedTime", new Timestamp(((Date)parameters[0]).getTime()));
205                break;
206            case GET_COORD_JOBS_OLDER_FOR_MATERILZATION:
207                query.setParameter("matTime", new Timestamp(((Date)parameters[0]).getTime()));
208                int limit = (Integer) parameters[1];
209                if (limit > 0) {
210                    query.setMaxResults(limit);
211                }
212                break;
213            case GET_COORD_FOR_ABANDONEDCHECK:
214                query.setParameter(1, (Integer) parameters[0]);
215                query.setParameter(2, (Timestamp) parameters[1]);
216                break;
217
218            case GET_COORD_IDS_FOR_STATUS_TRANSIT:
219                query.setParameter("lastModifiedTime", new Timestamp(((Date) parameters[0]).getTime()));
220                break;
221
222            default:
223                throw new JPAExecutorException(ErrorCode.E0603, "QueryExecutor cannot set parameters for "
224                        + namedQuery.name());
225        }
226        return query;
227    }
228
229    @Override
230    public int executeUpdate(CoordJobQuery namedQuery, CoordinatorJobBean jobBean) throws JPAExecutorException {
231        JPAService jpaService = Services.get().get(JPAService.class);
232        EntityManager em = jpaService.getEntityManager();
233        Query query = getUpdateQuery(namedQuery, jobBean, em);
234        int ret = jpaService.executeUpdate(namedQuery.name(), query, em);
235        return ret;
236    }
237
238    private CoordinatorJobBean constructBean(CoordJobQuery namedQuery, Object ret, Object... parameters)
239            throws JPAExecutorException {
240        CoordinatorJobBean bean;
241        Object[] arr;
242        switch (namedQuery) {
243            case GET_COORD_JOB:
244                bean = (CoordinatorJobBean) ret;
245                break;
246            case GET_COORD_JOB_USER_APPNAME:
247                bean = new CoordinatorJobBean();
248                arr = (Object[]) ret;
249                bean.setUser((String) arr[0]);
250                bean.setAppName((String) arr[1]);
251                break;
252            case GET_COORD_JOB_INPUT_CHECK:
253                bean = new CoordinatorJobBean();
254                arr = (Object[]) ret;
255                bean.setUser((String) arr[0]);
256                bean.setAppName((String) arr[1]);
257                bean.setStatusStr((String) arr[2]);
258                bean.setAppNamespace((String) arr[3]);
259                bean.setExecution((String) arr[4]);
260                bean.setFrequency((String) arr[5]);
261                bean.setTimeUnitStr((String) arr[6]);
262                bean.setTimeZone((String) arr[7]);
263                bean.setStartTime(DateUtils.toDate((Timestamp) arr[8]));
264                bean.setEndTime(DateUtils.toDate((Timestamp) arr[9]));
265                bean.setJobXmlBlob((StringBlob) arr[10]);
266                break;
267            case GET_COORD_JOB_ACTION_READY:
268                bean = new CoordinatorJobBean();
269                arr = (Object[]) ret;
270                bean.setId((String) arr[0]);
271                bean.setUser((String) arr[1]);
272                bean.setGroup((String) arr[2]);
273                bean.setAppName((String) arr[3]);
274                bean.setStatusStr((String) arr[4]);
275                bean.setExecution((String) arr[5]);
276                bean.setConcurrency((Integer) arr[6]);
277                bean.setFrequency((String) arr[7]);
278                bean.setTimeUnitStr((String) arr[8]);
279                bean.setTimeZone((String) arr[9]);
280                bean.setStartTime(DateUtils.toDate((Timestamp) arr[10]));
281                bean.setEndTime(DateUtils.toDate((Timestamp) arr[11]));
282                bean.setJobXmlBlob((StringBlob) arr[12]);
283                break;
284            case GET_COORD_JOB_ACTION_KILL:
285                bean = new CoordinatorJobBean();
286                arr = (Object[]) ret;
287                bean.setId((String) arr[0]);
288                bean.setUser((String) arr[1]);
289                bean.setGroup((String) arr[2]);
290                bean.setAppName((String) arr[3]);
291                bean.setStatusStr((String) arr[4]);
292                break;
293            case GET_COORD_JOB_MATERIALIZE:
294                bean = new CoordinatorJobBean();
295                arr = (Object[]) ret;
296                bean.setId((String) arr[0]);
297                bean.setUser((String) arr[1]);
298                bean.setGroup((String) arr[2]);
299                bean.setAppName((String) arr[3]);
300                bean.setStatusStr((String) arr[4]);
301                bean.setFrequency((String) arr[5]);
302                bean.setMatThrottling((Integer) arr[6]);
303                bean.setTimeout((Integer) arr[7]);
304                bean.setTimeZone((String) arr[8]);
305                bean.setStartTime(DateUtils.toDate((Timestamp) arr[9]));
306                bean.setEndTime(DateUtils.toDate((Timestamp) arr[10]));
307                bean.setPauseTime(DateUtils.toDate((Timestamp) arr[11]));
308                bean.setNextMaterializedTime(DateUtils.toDate((Timestamp) arr[12]));
309                bean.setLastActionTime(DateUtils.toDate((Timestamp) arr[13]));
310                bean.setLastActionNumber((Integer) arr[14]);
311                bean.setDoneMaterialization((Integer) arr[15]);
312                bean.setBundleId((String) arr[16]);
313                bean.setConfBlob((StringBlob) arr[17]);
314                bean.setJobXmlBlob((StringBlob) arr[18]);
315                bean.setAppNamespace((String) arr[19]);
316                bean.setTimeUnitStr((String) arr[20]);
317                bean.setExecution((String) arr[21]);
318                break;
319            case GET_COORD_JOB_SUSPEND_KILL:
320                bean = new CoordinatorJobBean();
321                arr = (Object[]) ret;
322                bean.setId((String) arr[0]);
323                bean.setUser((String) arr[1]);
324                bean.setGroup((String) arr[2]);
325                bean.setAppName((String) arr[3]);
326                bean.setStatusStr((String) arr[4]);
327                bean.setBundleId((String) arr[5]);
328                bean.setAppNamespace((String) arr[6]);
329                bean.setDoneMaterialization((Integer) arr[7]);
330                break;
331            case GET_COORD_JOB_STATUS:
332                bean = new CoordinatorJobBean();
333                bean.setId((String) parameters[0]);
334                bean.setStatusStr((String) ret);
335                break;
336            case GET_COORD_JOB_STATUS_PARENTID:
337                bean = new CoordinatorJobBean();
338                arr = (Object[]) ret;
339                bean.setId((String) parameters[0]);
340                bean.setStatusStr((String) arr[0]);
341                bean.setBundleId((String) arr[1]);
342                break;
343            case GET_COORD_JOBS_CHANGED:
344                bean = (CoordinatorJobBean) ret;
345                break;
346            case GET_COORD_JOBS_OLDER_FOR_MATERILZATION:
347                bean = new CoordinatorJobBean();
348                bean.setId((String) ret);
349                break;
350            case GET_COORD_FOR_ABANDONEDCHECK:
351                bean = new CoordinatorJobBean();
352                arr = (Object[]) ret;
353                bean.setId((String) arr[0]);
354                bean.setUser((String) arr[1]);
355                bean.setGroup((String) arr[2]);
356                bean.setAppName((String) arr[3]);
357                break;
358
359            case GET_COORD_IDS_FOR_STATUS_TRANSIT:
360                bean = new CoordinatorJobBean();
361                bean.setId((String) ret);
362                break;
363
364            default:
365                throw new JPAExecutorException(ErrorCode.E0603, "QueryExecutor cannot construct job bean for "
366                        + namedQuery.name());
367        }
368        return bean;
369    }
370
371    @Override
372    public CoordinatorJobBean get(CoordJobQuery namedQuery, Object... parameters) throws JPAExecutorException {
373        JPAService jpaService = Services.get().get(JPAService.class);
374        EntityManager em = jpaService.getEntityManager();
375        Query query = getSelectQuery(namedQuery, em, parameters);
376        Object ret = jpaService.executeGet(namedQuery.name(), query, em);
377        if (ret == null) {
378            throw new JPAExecutorException(ErrorCode.E0604, query.toString());
379        }
380        CoordinatorJobBean bean = constructBean(namedQuery, ret, parameters);
381        return bean;
382    }
383
384    @Override
385    public List<CoordinatorJobBean> getList(CoordJobQuery namedQuery, Object... parameters) throws JPAExecutorException {
386        JPAService jpaService = Services.get().get(JPAService.class);
387        EntityManager em = jpaService.getEntityManager();
388        Query query = getSelectQuery(namedQuery, em, parameters);
389        List<?> retList = (List<?>) jpaService.executeGetList(namedQuery.name(), query, em);
390        List<CoordinatorJobBean> beanList = new ArrayList<CoordinatorJobBean>();
391        if (retList != null) {
392            for (Object ret : retList) {
393                beanList.add(constructBean(namedQuery, ret, parameters));
394            }
395        }
396        return beanList;
397    }
398
399    @Override
400    public Object getSingleValue(CoordJobQuery namedQuery, Object... parameters) throws JPAExecutorException {
401        throw new UnsupportedOperationException();
402    }
403}