You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@oozie.apache.org by vi...@apache.org on 2013/04/18 01:55:21 UTC
svn commit: r1469108 [2/3] - in /oozie/trunk: ./ client/
client/src/main/java/org/apache/oozie/cli/
client/src/main/java/org/apache/oozie/client/
client/src/main/java/org/apache/oozie/client/event/jms/
client/src/main/java/org/apache/oozie/client/event...
Added: oozie/trunk/core/src/main/java/org/apache/oozie/executor/jpa/CoordinatorJobGetForUserJPAExecutor.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/executor/jpa/CoordinatorJobGetForUserJPAExecutor.java?rev=1469108&view=auto
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/executor/jpa/CoordinatorJobGetForUserJPAExecutor.java (added)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/executor/jpa/CoordinatorJobGetForUserJPAExecutor.java Wed Apr 17 23:55:20 2013
@@ -0,0 +1,67 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.oozie.executor.jpa;
+
+import javax.persistence.EntityManager;
+import javax.persistence.Query;
+
+import org.apache.oozie.ErrorCode;
+import org.apache.oozie.util.ParamChecker;
+
+/**
+ * Retrieve the name of user submitting coordinator
+ *
+ */
+public class CoordinatorJobGetForUserJPAExecutor implements JPAExecutor<String> {
+
+ private String wfJobId = null;
+
+ public CoordinatorJobGetForUserJPAExecutor(String wfJobId) {
+ ParamChecker.notNull(wfJobId, "wfJobId");
+ this.wfJobId = wfJobId;
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see org.apache.oozie.executor.jpa.JPAExecutor#getName()
+ */
+ @Override
+ public String getName() {
+ return "CoordinatorJobGetForUserJPAExecutor";
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see org.apache.oozie.executor.jpa.JPAExecutor#execute(javax.persistence.
+ * EntityManager)
+ */
+ @Override
+ public String execute(EntityManager em) throws JPAExecutorException {
+ try {
+ Query q = em.createNamedQuery("GET_COORD_JOB_FOR_USER");
+ q.setParameter("id", wfJobId);
+ String user = (String) q.getSingleResult();
+ return user;
+ }
+ catch (Exception e) {
+ throw new JPAExecutorException(ErrorCode.E0603, e.getMessage(), e);
+ }
+ }
+}
\ No newline at end of file
Added: oozie/trunk/core/src/main/java/org/apache/oozie/executor/jpa/WorkflowJobGetForUserJPAExecutor.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/executor/jpa/WorkflowJobGetForUserJPAExecutor.java?rev=1469108&view=auto
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/executor/jpa/WorkflowJobGetForUserJPAExecutor.java (added)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/executor/jpa/WorkflowJobGetForUserJPAExecutor.java Wed Apr 17 23:55:20 2013
@@ -0,0 +1,67 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.oozie.executor.jpa;
+
+import javax.persistence.EntityManager;
+import javax.persistence.Query;
+
+import org.apache.oozie.ErrorCode;
+import org.apache.oozie.util.ParamChecker;
+
+/**
+ *
+ * Retrieve the name of user submitting workflow
+ */
+public class WorkflowJobGetForUserJPAExecutor implements JPAExecutor<String> {
+
+ private String wfJobId = null;
+
+ public WorkflowJobGetForUserJPAExecutor(String wfJobId) {
+ ParamChecker.notNull(wfJobId, "wfJobId");
+ this.wfJobId = wfJobId;
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see org.apache.oozie.executor.jpa.JPAExecutor#getName()
+ */
+ @Override
+ public String getName() {
+ return "WorkflowJobGetForUserJPAExecutor";
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see org.apache.oozie.executor.jpa.JPAExecutor#execute(javax.persistence.
+ * EntityManager)
+ */
+ @Override
+ public String execute(EntityManager em) throws JPAExecutorException {
+ try {
+ Query q = em.createNamedQuery("GET_WORKFLOW_FOR_USER");
+ q.setParameter("id", wfJobId);
+ String user = (String) q.getSingleResult();
+ return user;
+ }
+ catch (Exception e) {
+ throw new JPAExecutorException(ErrorCode.E0603, e.getMessage(), e);
+ }
+ }
+}
\ No newline at end of file
Modified: oozie/trunk/core/src/main/java/org/apache/oozie/jms/ConnectionContext.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/jms/ConnectionContext.java?rev=1469108&r1=1469107&r2=1469108&view=diff
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/jms/ConnectionContext.java (original)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/jms/ConnectionContext.java Wed Apr 17 23:55:20 2013
@@ -27,31 +27,35 @@ import javax.jms.Session;
import javax.naming.NamingException;
/**
- * Maintains a JMS connection for creating session, consumer and producer
+ * Maintains a JMS connection for creating session, consumer and producer
*/
public interface ConnectionContext {
/**
* Create connection using properties
+ *
* @param props the properties used for creating jndi context
* @throws JMSException
*/
public void createConnection(Properties props) throws NamingException, JMSException;
/**
- * Set the exception Listener
- * @param exceptionListener
- */
+ * Set the exception Listener
+ *
+ * @param exceptionListener
+ */
public void setExceptionListener(ExceptionListener exceptionListener) throws JMSException;
/**
* Checks whether connection is initialized or not
+ *
* @return
*/
public boolean isConnectionInitialized();
/**
* Creates session using the specified session opts
+ *
* @param sessionOpts
* @return
* @throws JMSException
@@ -60,6 +64,7 @@ public interface ConnectionContext {
/**
* Creates consumer using session and topic name
+ *
* @param session
* @param topicName
* @return
@@ -68,7 +73,18 @@ public interface ConnectionContext {
public MessageConsumer createConsumer(Session session, String topicName) throws JMSException;
/**
+ * Creates consumer using session, topic name and selector
+ *
+ * @param session
+ * @param topicName
+ * @return
+ * @throws JMSException
+ */
+ public MessageConsumer createConsumer(Session session, String topicName, String selector) throws JMSException;
+
+ /**
* Creates producer using session and topic
+ *
* @param session
* @param topicName
* @return
@@ -77,6 +93,16 @@ public interface ConnectionContext {
public MessageProducer createProducer(Session session, String topicName) throws JMSException;
/**
+ * Creates a threadlocal session using session opts
+ *
+ * @param session
+ * @param topicName
+ * @return
+ * @throws JMSException
+ */
+ public ThreadLocal<Session> createThreadLocalSession(final int sessionOpts);
+
+ /**
* Closes the connection
*/
public void close();
Modified: oozie/trunk/core/src/main/java/org/apache/oozie/jms/DefaultConnectionContext.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/jms/DefaultConnectionContext.java?rev=1469108&r1=1469107&r2=1469108&view=diff
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/jms/DefaultConnectionContext.java (original)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/jms/DefaultConnectionContext.java Wed Apr 17 23:55:20 2013
@@ -38,6 +38,7 @@ public class DefaultConnectionContext im
protected Connection connection;
protected String connectionFactoryName;
private static XLog LOG = XLog.getLog(ConnectionContext.class);
+ private int sessionOpts;
@Override
public void createConnection(Properties props) throws NamingException, JMSException {
@@ -73,7 +74,6 @@ public class DefaultConnectionContext im
}
}
-
@Override
public boolean isConnectionInitialized() {
return connection != null;
@@ -113,4 +113,29 @@ public class DefaultConnectionContext im
}
}
+ private final ThreadLocal<Session> th = new ThreadLocal<Session>() {
+ protected Session initialValue() {
+ try {
+ return connection.createSession(false, sessionOpts);
+ }
+ catch (JMSException e) {
+ throw new RuntimeException("Session couldn't be created", e);
+ }
+ }
+ };
+
+ @Override
+ public ThreadLocal<Session> createThreadLocalSession(final int sessionOpts) {
+ this.sessionOpts = sessionOpts;
+ return th;
+ }
+
+
+ @Override
+ public MessageConsumer createConsumer(Session session, String topicName, String selector) throws JMSException {
+ Topic topic = session.createTopic(topicName);
+ MessageConsumer consumer = session.createConsumer(topic, selector);
+ return consumer;
+ }
+
}
Added: oozie/trunk/core/src/main/java/org/apache/oozie/jms/DefaultJMSServerInfo.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/jms/DefaultJMSServerInfo.java?rev=1469108&view=auto
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/jms/DefaultJMSServerInfo.java (added)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/jms/DefaultJMSServerInfo.java Wed Apr 17 23:55:20 2013
@@ -0,0 +1,46 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.oozie.jms;
+
+import org.apache.oozie.client.rest.JMSConnectionInfoBean;
+import org.apache.oozie.service.JMSTopicService;
+import org.apache.oozie.service.Services;
+
+/**
+ * Default implementation to retrieve the JMS ConnectionInfo bean
+ */
+public class DefaultJMSServerInfo implements JMSServerInfo {
+ private JMSTopicService jmsTopicService = Services.get().get(JMSTopicService.class);
+
+ @Override
+ public JMSConnectionInfoBean getJMSConnectionInfoBean(String connectionProperties, String jobId) throws Exception {
+ JMSConnectionInfoBean jmsBean = new JMSConnectionInfoBean();
+ JMSConnectionInfo jmsInfo = new JMSConnectionInfo(connectionProperties);
+ jmsBean.setJNDIProperties(jmsInfo.getJNDIProperties());
+ if (jmsTopicService != null) {
+ jmsBean.setTopicName(jmsTopicService.getTopic(jobId));
+ }
+ else {
+ throw new Exception(
+ "Topic name cannot be retrieved as JMSTopicService is not initialized. JMS notification"
+ + "may not be enabled");
+ }
+ return jmsBean;
+ }
+
+}
Modified: oozie/trunk/core/src/main/java/org/apache/oozie/jms/JMSExceptionListener.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/jms/JMSExceptionListener.java?rev=1469108&r1=1469107&r2=1469108&view=diff
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/jms/JMSExceptionListener.java (original)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/jms/JMSExceptionListener.java Wed Apr 17 23:55:20 2013
@@ -29,6 +29,7 @@ public class JMSExceptionListener implem
private static XLog LOG = XLog.getLog(JMSExceptionListener.class);
private JMSConnectionInfo connInfo;
private ConnectionContext connCtxt;
+ private boolean retry;
/**
* Create ExceptionLister for a JMS Connection
@@ -36,17 +37,21 @@ public class JMSExceptionListener implem
* @param connInfo Information to connect to the JMS compliant messaging service
* @param connCtxt The actual connection on which this listener will be registered
*/
- public JMSExceptionListener(JMSConnectionInfo connInfo, ConnectionContext connCtxt) {
+ public JMSExceptionListener(JMSConnectionInfo connInfo, ConnectionContext connCtxt, boolean retry) {
this.connInfo = connInfo;
this.connCtxt = connCtxt;
+ this.retry = retry;
}
@Override
public void onException(JMSException exception) {
- LOG.warn("Received JMSException for [{0}]. Reestablishing connection", connInfo, exception);
+ LOG.warn("Received JMSException for [{0}]", connInfo, exception);
connCtxt.close();
JMSAccessorService jmsService = Services.get().get(JMSAccessorService.class);
- jmsService.reestablishConnection(connInfo);
+ jmsService.removeConnInfo(connInfo);
+ if (retry) {
+ jmsService.reestablishConnection(connInfo);
+ }
}
}
Added: oozie/trunk/core/src/main/java/org/apache/oozie/jms/JMSJobEventListener.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/jms/JMSJobEventListener.java?rev=1469108&view=auto
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/jms/JMSJobEventListener.java (added)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/jms/JMSJobEventListener.java Wed Apr 17 23:55:20 2013
@@ -0,0 +1,166 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.oozie.jms;
+
+import java.util.Map;
+
+import javax.jms.DeliveryMode;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.oozie.client.event.jms.JMSHeaderConstants;
+import org.apache.oozie.client.event.message.CoordinatorActionMessage;
+import org.apache.oozie.client.event.message.JobMessage;
+import org.apache.oozie.client.event.message.WorkflowJobMessage;
+import org.apache.oozie.event.BundleJobEvent;
+import org.apache.oozie.event.CoordinatorActionEvent;
+import org.apache.oozie.event.CoordinatorJobEvent;
+import org.apache.oozie.event.WorkflowActionEvent;
+import org.apache.oozie.event.WorkflowJobEvent;
+import org.apache.oozie.event.listener.JobEventListener;
+import org.apache.oozie.event.messaging.MessageFactory;
+import org.apache.oozie.event.messaging.MessageSerializer;
+import org.apache.oozie.service.JMSAccessorService;
+import org.apache.oozie.service.JMSTopicService;
+import org.apache.oozie.service.Services;
+import org.apache.oozie.util.XLog;
+
+/**
+ * Class to send JMS notifications related to job events.
+ *
+ */
+public class JMSJobEventListener extends JobEventListener {
+ private JMSAccessorService jmsService = Services.get().get(JMSAccessorService.class);
+ private JMSTopicService jmsTopicService = Services.get().get(JMSTopicService.class);
+ private JMSConnectionInfo connInfo;
+ public static final String JMS_CONNECTION_PROPERTIES = "oozie.jms.producer.notification.connection";
+ public static final String JMS_SESSION_OPTS = "oozie.jms.producer.session.opts";
+ public static final String JMS_DELIVERY_MODE = "oozie.jms.delivery.mode";
+ public static final String JMS_EXPIRATION_DATE = "oozie.jms.expiration.date";
+ private int jmsSessionOpts;
+ private int jmsDeliveryMode;
+ private int jmsExpirationDate;
+ private ConnectionContext jmsContext;
+ private static XLog LOG;
+
+ @Override
+ public void init() {
+ LOG = XLog.getLog(getClass());
+ Configuration conf = Services.get().getConf();
+ String jmsProps = conf.get(JMS_CONNECTION_PROPERTIES);
+ LOG.info("JMS producer connection properties [{0}]", jmsProps);
+ connInfo = new JMSConnectionInfo(jmsProps);
+ jmsSessionOpts = conf.getInt(JMS_SESSION_OPTS, Session.AUTO_ACKNOWLEDGE);
+ jmsDeliveryMode = conf.getInt(JMS_DELIVERY_MODE, DeliveryMode.PERSISTENT);
+ jmsExpirationDate = conf.getInt(JMS_EXPIRATION_DATE, 0);
+
+ }
+
+ protected void sendMessage(Map<String, String> messageProperties, String messageBody, String topicName,
+ String messageFormat) {
+ jmsContext = jmsService.createConnectionContext(connInfo, false);
+ ThreadLocal<Session> threadLocal = null;
+ try {
+ threadLocal = jmsContext.createThreadLocalSession(jmsSessionOpts);
+ Session session = threadLocal.get();
+ TextMessage textMessage = session.createTextMessage(messageBody);
+ for (Map.Entry<String, String> property : messageProperties.entrySet()) {
+ textMessage.setStringProperty(property.getKey(), property.getValue());
+ }
+ textMessage.setStringProperty(JMSHeaderConstants.MESSAGE_FORMAT, messageFormat);
+ LOG.trace("Event related JMS message [{0}]", textMessage.getText());
+ MessageProducer producer = jmsContext.createProducer(session, topicName);
+ producer.setDeliveryMode(jmsDeliveryMode);
+ producer.setTimeToLive(jmsExpirationDate);
+ producer.send(textMessage);
+ producer.close();
+ }
+ catch (Exception jmse) {
+ LOG.error("Exception happened while sending event related jms message", jmse);
+ if (threadLocal != null) {
+ threadLocal.remove();
+ }
+ }
+
+ }
+
+ @Override
+ public void onWorkflowJobEvent(WorkflowJobEvent event) {
+ WorkflowJobMessage wfJobMessage = MessageFactory.createWorkflowJobMessage(event);
+ serializeAndSendJMSMessage(wfJobMessage, getTopic(event));
+
+ }
+
+ @Override
+ public void onCoordinatorActionEvent(CoordinatorActionEvent event) {
+ CoordinatorActionMessage coordActionMessage = MessageFactory.createCoordinatorActionMessage(event);
+ serializeAndSendJMSMessage(coordActionMessage, getTopic(event));
+ }
+
+ private void serializeAndSendJMSMessage(JobMessage jobMessage, String topicName) {
+ MessageSerializer serializer = MessageFactory.getMessageSerializer();
+ String messageBody = serializer.getSerializedObject(jobMessage);
+ sendMessage(jobMessage.getMessageProperties(), messageBody, topicName, serializer.getMessageFormat());
+ }
+
+ protected String getTopic(WorkflowJobEvent event) {
+ if (jmsTopicService != null) {
+ return jmsTopicService.getTopic(event);
+ }
+ else {
+ throw new RuntimeException("JMSTopicService is not initialized");
+ }
+ }
+
+ protected String getTopic(CoordinatorActionEvent event) {
+ if (jmsTopicService != null) {
+ return jmsTopicService.getTopic(event);
+ }
+ else {
+ throw new RuntimeException("JMSTopicService is not initialized");
+ }
+ }
+
+ @Override
+ public void onWorkflowActionEvent(WorkflowActionEvent wae) {
+ LOG.warn("Sending jms message related to workflow action is not supported");
+ }
+
+ @Override
+ public void onCoordinatorJobEvent(CoordinatorJobEvent wje) {
+ LOG.warn("Sending jms message related to coordinator job is not supported");
+ }
+
+ @Override
+ public void onBundleJobEvent(BundleJobEvent wje) {
+ LOG.warn("Sending jms message related to bundle job is not supported");
+ }
+
+ @Override
+ public void destroy() {
+ if (jmsContext != null) {
+ LOG.debug("Closing JMS connection");
+ jmsContext.close();
+ jmsService.removeConnInfo(connInfo);
+ }
+
+ }
+
+}
Added: oozie/trunk/core/src/main/java/org/apache/oozie/jms/JMSServerInfo.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/jms/JMSServerInfo.java?rev=1469108&view=auto
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/jms/JMSServerInfo.java (added)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/jms/JMSServerInfo.java Wed Apr 17 23:55:20 2013
@@ -0,0 +1,37 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.oozie.jms;
+
+import org.apache.oozie.client.rest.JMSConnectionInfoBean;
+import org.apache.oozie.executor.jpa.JPAExecutorException;
+
+/**
+ * Get the JMS ConnectionInfoBean
+ */
+public interface JMSServerInfo {
+
+ /**
+ * Retrive the conn info bean using conn properties and job id
+ * @param connectionProperties the jms producer conn properties
+ * @param jobId the job id
+ * @return
+ * @throws Exception
+ */
+ public JMSConnectionInfoBean getJMSConnectionInfoBean(String connectionProperties, String jobId) throws Exception;
+
+}
Modified: oozie/trunk/core/src/main/java/org/apache/oozie/service/EventHandlerService.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/service/EventHandlerService.java?rev=1469108&r1=1469107&r2=1469108&view=diff
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/service/EventHandlerService.java (original)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/service/EventHandlerService.java Wed Apr 17 23:55:20 2013
@@ -105,7 +105,7 @@ public class EventHandlerService impleme
}
private void initEventListeners(Configuration conf) {
- Class<?>[] listenerClass = (Class<?>[]) conf.getClasses(CONF_LISTENERS);
+ Class<?>[] listenerClass = (Class<?>[]) conf.getClasses(CONF_LISTENERS, org.apache.oozie.jms.JMSJobEventListener.class);
for (int i = 0; i < listenerClass.length; i++) {
Object listener = null;
try {
@@ -170,7 +170,6 @@ public class EventHandlerService impleme
}
}
}
- eventsConfigured = false;
}
@Override
@@ -231,19 +230,19 @@ public class EventHandlerService impleme
JobEventListener el = (JobEventListener) iter.next();
switch (event.getAppType()) {
case WORKFLOW_JOB:
- onWorkflowJobEvent(event, el);
+ el.onWorkflowJobEvent((WorkflowJobEvent)event);
break;
case WORKFLOW_ACTION:
- onWorkflowActionEvent(event, el);
+ el.onWorkflowActionEvent((WorkflowActionEvent)event);
break;
case COORDINATOR_JOB:
- onCoordinatorJobEvent(event, el);
+ el.onCoordinatorJobEvent((CoordinatorJobEvent)event);
break;
case COORDINATOR_ACTION:
- onCoordinatorActionEvent(event, el);
+ el.onCoordinatorActionEvent((CoordinatorActionEvent)event);
break;
case BUNDLE_JOB:
- onBundleJobEvent(event, el);
+ el.onBundleJobEvent((BundleJobEvent)event);
break;
default:
XLog.getLog(EventHandlerService.class).info("Undefined Job Event app-type - {0}",
@@ -276,107 +275,6 @@ public class EventHandlerService impleme
XLog.getLog(EventHandlerService.class).info("Undefined SLA event type - {0}", event.getEventType());
}
}
-
- private void onWorkflowJobEvent(JobEvent event, JobEventListener el) {
- switch (event.getEventStatus()) {
- case STARTED:
- el.onWorkflowJobStart((WorkflowJobEvent) event);
- break;
- case SUCCESS:
- el.onWorkflowJobSuccess((WorkflowJobEvent) event);
- break;
- case FAILURE:
- el.onWorkflowJobFailure((WorkflowJobEvent) event);
- break;
- case SUSPEND:
- el.onWorkflowJobSuspend((WorkflowJobEvent) event);
- break;
- default:
- XLog.getLog(EventHandlerService.class).info("Undefined WF Job event-status - {0}",
- event.getEventStatus());
- }
- }
-
- private void onWorkflowActionEvent(JobEvent event, JobEventListener el) {
- switch (event.getEventStatus()) {
- case STARTED:
- el.onWorkflowActionStart((WorkflowActionEvent) event);
- break;
- case SUCCESS:
- el.onWorkflowActionSuccess((WorkflowActionEvent) event);
- break;
- case FAILURE:
- el.onWorkflowActionFailure((WorkflowActionEvent) event);
- break;
- default:
- XLog.getLog(EventHandlerService.class).info("Undefined WF action event-status - {0}",
- event.getEventStatus());
- }
- }
-
- private void onCoordinatorJobEvent(JobEvent event, JobEventListener el) {
- switch (event.getEventStatus()) {
- case STARTED:
- el.onCoordinatorJobStart((CoordinatorJobEvent) event);
- break;
- case SUCCESS:
- el.onCoordinatorJobSuccess((CoordinatorJobEvent) event);
- break;
- case FAILURE:
- el.onCoordinatorJobFailure((CoordinatorJobEvent) event);
- break;
- case SUSPEND:
- el.onCoordinatorJobSuspend((CoordinatorJobEvent) event);
- break;
- default:
- XLog.getLog(EventHandlerService.class).info("Undefined Coord Job event-status - {0}",
- event.getEventStatus());
- }
- }
-
- private void onCoordinatorActionEvent(JobEvent event, JobEventListener el) {
- switch (event.getEventStatus()) {
- case WAITING:
- el.onCoordinatorActionWaiting((CoordinatorActionEvent) event);
- break;
- case STARTED:
- el.onCoordinatorActionStart((CoordinatorActionEvent) event);
- break;
- case SUCCESS:
- el.onCoordinatorActionSuccess((CoordinatorActionEvent) event);
- break;
- case FAILURE:
- el.onCoordinatorActionFailure((CoordinatorActionEvent) event);
- break;
- case SUSPEND:
- el.onCoordinatorActionSuspend((CoordinatorActionEvent) event);
- break;
- default:
- XLog.getLog(EventHandlerService.class).info("Undefined Coord action event-status - {0}",
- event.getEventStatus());
- }
- }
-
- private void onBundleJobEvent(JobEvent event, JobEventListener el) {
- switch (event.getEventStatus()) {
- case STARTED:
- el.onBundleJobStart((BundleJobEvent) event);
- break;
- case SUCCESS:
- el.onBundleJobSuccess((BundleJobEvent) event);
- break;
- case FAILURE:
- el.onBundleJobFailure((BundleJobEvent) event);
- break;
- case SUSPEND:
- el.onBundleJobSuspend((BundleJobEvent) event);
- break;
- default:
- XLog.getLog(EventHandlerService.class).info("Undefined Bundle Job event-type - {0}",
- event.getEventStatus());
- }
- }
-
}
}
Modified: oozie/trunk/core/src/main/java/org/apache/oozie/service/JMSAccessorService.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/service/JMSAccessorService.java?rev=1469108&r1=1469107&r2=1469108&view=diff
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/service/JMSAccessorService.java (original)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/service/JMSAccessorService.java Wed Apr 17 23:55:20 2013
@@ -54,6 +54,7 @@ public class JMSAccessorService implemen
public static final String CONF_RETRY_INITIAL_DELAY = CONF_PREFIX + "retry.initial.delay";
public static final String CONF_RETRY_MULTIPLIER = CONF_PREFIX + "retry.multiplier";
public static final String CONF_RETRY_MAX_ATTEMPTS = CONF_PREFIX + "retry.max.attempts";
+ public static final String JMS_PRODUCER_CONNECTION_INFO_IMPL = CONF_PREFIX + "jms.producer.connection.info.impl";
private static XLog LOG;
private Configuration conf;
@@ -105,7 +106,7 @@ public class JMSAccessorService implemen
if (!topicsMap.containsKey(topic)) {
synchronized (topicsMap) {
if (!topicsMap.containsKey(topic)) {
- ConnectionContext connCtxt = createConnectionContext(connInfo);
+ ConnectionContext connCtxt = createConnectionContext(connInfo, true);
if (connCtxt == null) {
queueTopicForRetry(connInfo, topic, msgHandler);
return;
@@ -254,13 +255,13 @@ public class JMSAccessorService implemen
}
}
- protected ConnectionContext createConnectionContext(JMSConnectionInfo connInfo) {
+ public ConnectionContext createConnectionContext(JMSConnectionInfo connInfo, boolean retry) {
ConnectionContext connCtxt = connectionMap.get(connInfo);
if (connCtxt == null) {
try {
connCtxt = getConnectionContextImpl();
connCtxt.createConnection(connInfo.getJNDIProperties());
- connCtxt.setExceptionListener(new JMSExceptionListener(connInfo, connCtxt));
+ connCtxt.setExceptionListener(new JMSExceptionListener(connInfo, connCtxt, retry));
connectionMap.put(connInfo, connCtxt);
LOG.info("Connection established to JMS Server for [{0}]", connInfo);
}
@@ -321,12 +322,15 @@ public class JMSAccessorService implemen
return JMSAccessorService.class;
}
+ public void removeConnInfo(JMSConnectionInfo connInfo){
+ connectionMap.remove(connInfo);
+ }
+
/**
* Reestablish connection for the given JMS connect information
* @param connInfo JMS connection info
*/
public void reestablishConnection(JMSConnectionInfo connInfo) {
- connectionMap.remove(connInfo);
// Queue the connection and topics for retry
ConnectionRetryInfo connRetryInfo = queueConnectionForRetry(connInfo);
Map<String, MessageReceiver> listeningTopicsMap = receiversMap.remove(connInfo);
@@ -362,7 +366,7 @@ public class JMSAccessorService implemen
LOG.info("Attempting retry of connection [{0}]", connInfo);
connRetryInfo.setNumAttempt(connRetryInfo.getNumAttempt() + 1);
connRetryInfo.setNextDelay(connRetryInfo.getNextDelay() * retryMultiplier);
- ConnectionContext connCtxt = createConnectionContext(connInfo);
+ ConnectionContext connCtxt = createConnectionContext(connInfo, true);
boolean shouldRetry = false;
if (connCtxt == null) {
shouldRetry = true;
Added: oozie/trunk/core/src/main/java/org/apache/oozie/service/JMSTopicService.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/service/JMSTopicService.java?rev=1469108&view=auto
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/service/JMSTopicService.java (added)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/service/JMSTopicService.java Wed Apr 17 23:55:20 2013
@@ -0,0 +1,266 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.oozie.service;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.oozie.ErrorCode;
+import org.apache.oozie.event.CoordinatorActionEvent;
+import org.apache.oozie.event.WorkflowJobEvent;
+import org.apache.oozie.executor.jpa.BundleJobGetForUserJPAExecutor;
+import org.apache.oozie.executor.jpa.CoordinatorJobGetForUserJPAExecutor;
+import org.apache.oozie.executor.jpa.JPAExecutorException;
+import org.apache.oozie.executor.jpa.WorkflowJobGetForUserJPAExecutor;
+import org.apache.oozie.util.XLog;
+
+/**
+ * JMS Topic service to retrieve topic names from events or job id
+ *
+ */
+public class JMSTopicService implements Service {
+
+ public static final String CONF_PREFIX = Service.CONF_PREFIX + "JMSTopicService.";
+ public static final String TOPIC_NAME = CONF_PREFIX + "topic.name";
+ public static final String TOPIC_PREFIX = CONF_PREFIX + "topic.prefix";
+ private static XLog LOG;
+ private Configuration conf;
+ private final Map<String, String> topicMap = new HashMap<String, String>();
+ private static final List<String> JOB_TYPE_CONSTANTS = new ArrayList<String>();
+ private static final List<String> ALLOWED_TOPIC_NAMES = new ArrayList<String>();
+ private JPAService jpaService = Services.get().get(JPAService.class);
+ private String defaultTopicName = TopicType.USER.value;
+ private String topicPrefix;
+
+ static {
+ ALLOWED_TOPIC_NAMES.add(TopicType.USER.value);
+ ALLOWED_TOPIC_NAMES.add(TopicType.JOBID.value);
+ }
+
+ static {
+ JOB_TYPE_CONSTANTS.add(JobType.WORKFLOW.value);
+ JOB_TYPE_CONSTANTS.add(JobType.COORDINATOR.value);
+ JOB_TYPE_CONSTANTS.add(JobType.BUNDLE.value);
+ }
+
+ public static enum JobType {
+ WORKFLOW("WORKFLOW"), COORDINATOR("COORDINATOR"), BUNDLE("BUNDLE");
+ private String value;
+
+ JobType(String value) {
+ this.value = value;
+ }
+
+ String getValue() {
+ return value;
+ }
+
+ }
+
+ public static enum TopicType {
+ USER("${username}"), JOBID("${jobId}");
+
+ private String value;
+
+ TopicType(String value) {
+ this.value = value;
+ }
+
+ String getValue() {
+ return value;
+ }
+
+ }
+
+ @Override
+ public void init(Services services) throws ServiceException {
+ LOG = XLog.getLog(getClass());
+ conf = services.getConf();
+ parseTopicConfiguration();
+ topicPrefix = conf.get(TOPIC_PREFIX, "");
+ }
+
+ private void parseTopicConfiguration() throws ServiceException {
+ String topicName = conf.get(TOPIC_NAME, "default=" + TopicType.USER.value);
+ if (topicName == null) {
+ throw new ServiceException(ErrorCode.E0100, getClass().getName(), "JMS topic cannot be null ");
+ }
+ LOG.info("Topic Name is [{0}]", topicName);
+ String[] topic = topicName.trim().split(",");
+ for (int i = 0; i < topic.length; i++) {
+ String[] split = topic[i].trim().split("=");
+ if (split.length == 2) {
+ split[0] = split[0].trim();
+ split[1] = split[1].trim();
+ if (split[0].equals("default")) {
+ if (!ALLOWED_TOPIC_NAMES.contains(split[1])) {
+ throw new ServiceException(ErrorCode.E0100, getClass().getName(), "Topic name " + split[1]
+ + " not allowed in default; allowed" + "topics are " + ALLOWED_TOPIC_NAMES);
+ }
+ defaultTopicName = split[1];
+ }
+ else if (!JOB_TYPE_CONSTANTS.contains(split[0])) {
+ throw new ServiceException(ErrorCode.E0100, getClass().getName(),
+ "Incorrect job type for defining JMS topic: " + split[0] + " ;" + "allowed job types are "
+ + JOB_TYPE_CONSTANTS);
+ }
+ else if (!ALLOWED_TOPIC_NAMES.contains(split[1]) && split[1].contains("$")) {
+ throw new ServiceException(ErrorCode.E0100, getClass().getName(), "JMS topic value " + split[1]
+ + " " + "for a job type is incorrect " + "Correct values are " + ALLOWED_TOPIC_NAMES);
+ }
+ else {
+ topicMap.put(split[0], split[1]);
+ }
+ }
+ else {
+ throw new ServiceException(ErrorCode.E0100, getClass().getName(), "Property " + topic[i]
+ + "has incorrect syntax; It should be specified as key value pair");
+ }
+ }
+ }
+
+ /**
+ * Retrieve topic from Job id
+ * @param jobId
+ * @return
+ * @throws JPAExecutorException
+ */
+ public String getTopic(String jobId) throws JPAExecutorException {
+ String topicName = null;
+ if (jobId.contains("-W@")) {
+ jobId = jobId.substring(0, jobId.indexOf('@'));
+ topicName = getTopicForWorkflow(jobId);
+ }
+ else if (jobId.endsWith("-W")) {
+ topicName = getTopicForWorkflow(jobId);
+ }
+ else if (jobId.contains("-C@")) {
+ jobId = jobId.substring(0, jobId.indexOf('@'));
+ topicName = getTopicForCoordinator(jobId);
+ }
+ else if (jobId.endsWith("C")) {
+ topicName = getTopicForCoordinator(jobId);
+ }
+ else if (jobId.contains("-B_")) {
+ jobId = jobId.substring(0, jobId.indexOf('_'));
+ topicName = getTopicForBundle(jobId);
+ }
+
+ else if (jobId.endsWith("B")) {
+ topicName = getTopicForBundle(jobId);
+ }
+ return topicPrefix + topicName;
+ }
+
+ /**
+ * Retrieve topic from coord event
+ * @param event the coordinatorAction event
+ * @return
+ */
+ public String getTopic(CoordinatorActionEvent event) {
+ String topicName = topicMap.get(JobType.COORDINATOR.value);
+ if (topicName == null) {
+ if (defaultTopicName.equals(TopicType.USER.value)) {
+ topicName = event.getUser();
+ }
+ else if (defaultTopicName.equals(TopicType.JOBID.value)) {
+ topicName = event.getParentId();
+ }
+ }
+ return topicPrefix + topicName;
+
+ }
+
+ /**
+ * Retrieve topic for workflow job event
+ * @param event the wf job event
+ * @return
+ */
+ public String getTopic(WorkflowJobEvent event) {
+ String topicName = topicMap.get(JobType.WORKFLOW.value);
+ if (topicName == null) {
+ if (defaultTopicName.equals(TopicType.USER.value)) {
+ topicName = event.getUser();
+ }
+ else if (defaultTopicName.equals(TopicType.JOBID.value)) {
+ topicName = event.getId();
+ }
+ }
+ return topicPrefix + topicName;
+
+ }
+
+ private String getTopicForWorkflow(String jobId) throws JPAExecutorException {
+ String topicName = topicMap.get(JobType.WORKFLOW.value);
+ if (topicName == null) {
+ topicName = defaultTopicName;
+ }
+ if (topicName.equals(TopicType.USER.value)) {
+ topicName = jpaService.execute(new WorkflowJobGetForUserJPAExecutor(jobId));
+ }
+ else if (topicName.equals(TopicType.JOBID.value)) {
+ topicName = jobId;
+ }
+ return topicName;
+
+ }
+
+ private String getTopicForCoordinator(String jobId) throws JPAExecutorException {
+ String topicName = topicMap.get(JobType.COORDINATOR.value);
+
+ if (topicName == null) {
+ topicName = defaultTopicName;
+ }
+ if (topicName.equals(TopicType.USER.value)) {
+ topicName = jpaService.execute(new CoordinatorJobGetForUserJPAExecutor(jobId));
+ }
+ else if (topicName.equals(TopicType.JOBID.value)) {
+ topicName = jobId;
+ }
+ return topicName;
+ }
+
+ private String getTopicForBundle(String jobId) throws JPAExecutorException {
+ String topicName = topicMap.get(JobType.BUNDLE.value);
+ if (topicName == null) {
+ topicName = defaultTopicName;
+ }
+ if (topicName.equals(TopicType.USER.value)) {
+ topicName = jpaService.execute(new BundleJobGetForUserJPAExecutor(jobId));
+ }
+ else if (topicName.equals(TopicType.JOBID.value)) {
+ topicName = jobId;
+ }
+ return topicName;
+ }
+
+ @Override
+ public void destroy() {
+ topicMap.clear();
+
+ }
+
+ @Override
+ public Class<? extends Service> getInterface() {
+ return JMSTopicService.class;
+ }
+
+}
Modified: oozie/trunk/core/src/main/java/org/apache/oozie/servlet/BaseJobServlet.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/servlet/BaseJobServlet.java?rev=1469108&r1=1469107&r2=1469108&view=diff
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/servlet/BaseJobServlet.java (original)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/servlet/BaseJobServlet.java Wed Apr 17 23:55:20 2013
@@ -6,9 +6,9 @@
* to you 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.
@@ -235,6 +235,14 @@ public abstract class BaseJobServlet ext
startCron();
sendJsonResponse(response, HttpServletResponse.SC_OK, job, timeZoneId);
}
+
+ else if (show.equals(RestConstants.JOB_SHOW_JMS_INFO)){
+ stopCron();
+ JsonBean jmsBean = getJMSConnectionInfo (request, response);
+ startCron();
+ sendJsonResponse(response, HttpServletResponse.SC_OK, jmsBean, timeZoneId);
+ }
+
else if (show.equals(RestConstants.JOB_SHOW_LOG)) {
response.setContentType(TEXT_UTF8);
streamJobLog(request, response);
@@ -371,4 +379,15 @@ public abstract class BaseJobServlet ext
*/
abstract void streamJobGraph(HttpServletRequest request, HttpServletResponse response)
throws XServletException, IOException;
+
+
+ /**
+ * abstract method to get JMS connection details for a job
+ * @param request
+ * @param response
+ * @throws XServletException
+ * @throws IOException
+ */
+ abstract JsonBean getJMSConnectionInfo(HttpServletRequest request, HttpServletResponse response)
+ throws XServletException, IOException;
}
Modified: oozie/trunk/core/src/main/java/org/apache/oozie/servlet/V0JobServlet.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/servlet/V0JobServlet.java?rev=1469108&r1=1469107&r2=1469108&view=diff
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/servlet/V0JobServlet.java (original)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/servlet/V0JobServlet.java Wed Apr 17 23:55:20 2013
@@ -6,9 +6,9 @@
* to you 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.
@@ -209,4 +209,9 @@ public class V0JobServlet extends BaseJo
// Should this error code be NOT_IMPLEMENTED?
throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ErrorCode.E0306);
}
+
+ @Override
+ protected JsonBean getJMSConnectionInfo(HttpServletRequest request, HttpServletResponse response)
+ throws XServletException, IOException {
+ throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ErrorCode.E0306);}
}
\ No newline at end of file
Modified: oozie/trunk/core/src/main/java/org/apache/oozie/servlet/V1JobServlet.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/servlet/V1JobServlet.java?rev=1469108&r1=1469107&r2=1469108&view=diff
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/servlet/V1JobServlet.java (original)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/servlet/V1JobServlet.java Wed Apr 17 23:55:20 2013
@@ -685,6 +685,20 @@ public class V1JobServlet extends BaseJo
return jobBean;
}
+ protected JsonBean getJMSConnectionInfo(HttpServletRequest request, HttpServletResponse response) throws XServletException{
+ JsonBean jmsBean = null;
+ String jobId = getResourceName(request);
+ DagEngine dagEngine = Services.get().get(DagEngineService.class).getDagEngine(getUser(request),
+ getAuthToken(request));
+ try {
+ jmsBean = (JsonBean) dagEngine.getJMSConnectionInfo(jobId);
+ }
+ catch (DagEngineException ex) {
+ throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ex);
+ }
+ return jmsBean;
+ }
+
/**
* Get wf action info
*
Modified: oozie/trunk/core/src/main/resources/oozie-default.xml
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/resources/oozie-default.xml?rev=1469108&r1=1469107&r2=1469108&view=diff
==============================================================================
--- oozie/trunk/core/src/main/resources/oozie-default.xml (original)
+++ oozie/trunk/core/src/main/resources/oozie-default.xml Wed Apr 17 23:55:20 2013
@@ -145,6 +145,34 @@
Mapping rules can also be provided for mapping Hcatalog servers to corresponding JMS providers.
hcat://${1}.${2}.server.com:8020=java.naming.factory.initial#Dummy.Factory;java.naming.provider.url#tcp://broker.${2}:61616
</description>
+ </property>
+
+ <!-- TopicService -->
+
+ <property>
+ <name>oozie.service.JMSTopicService.topic.name</name>
+ <value>
+ default=${username}
+ </value>
+ <description>
+ Other topic options are ${jobid} or a fixed string.
+ A fixed string can also be specified as topic for a particular job type.
+ For e.g To have a fixed string topic for workflows, coordinators and bundles,
+ specify in the following comma-separated format: {jobtype1}={some_string1}, {jobtype2}={some_string2}
+ where job type can be WORKFLOW, COORDINATOR or BUNDLE
+ E.g Following defines topic for workflow job, workflow action, coordinator job, coordinator action,
+ bundle job and bundle action
+ WORKFLOW=workflow,
+ COORDINATOR=coordinator,
+ BUNDLE={jobId}
+ For jobs with no defined topic, default topic will be ${username}
+ </description>
+ </property>
+
+ <!-- JMS Producer connection -->
+ <property>
+ <name>oozie.jms.producer.notification.connection</name>
+ <value>java.naming.factory.initial#org.apache.activemq.jndi.ActiveMQInitialContextFactory;java.naming.provider.url#tcp://localhost:61616;connectionFactoryNames#ConnectionFactory</value>
</property>
<!-- JMSAccessorService -->
@@ -1367,7 +1395,7 @@
<name>oozie.service.StatusTransitService.backward.support.for.coord.status</name>
<value>false</value>
<description>
- true, if coordinator job submits using 'uri:oozie:coordinator:0.1' namespace and wants to keep Oozie 2.x status transit.
+ true, if coordinator job submits using 'uri:oozie:coordinator:0.1' and wants to keep Oozie 2.x status transit.
if set true,
1. SUCCEEDED state in coordinator job means materialization done.
2. No DONEWITHERROR state in coordinator job
@@ -1803,6 +1831,11 @@
</property>
<property>
+ <name>oozie.service.EventHandlerService.event.listeners</name>
+ <value>org.apache.oozie.jms.JMSJobEventListener</value>
+ </property>
+
+ <property>
<name>oozie.service.EventHandlerService.queue.size</name>
<value>10000</value>
<description>
Added: oozie/trunk/core/src/test/java/org/apache/oozie/command/TestJMSInfoXCommand.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/test/java/org/apache/oozie/command/TestJMSInfoXCommand.java?rev=1469108&view=auto
==============================================================================
--- oozie/trunk/core/src/test/java/org/apache/oozie/command/TestJMSInfoXCommand.java (added)
+++ oozie/trunk/core/src/test/java/org/apache/oozie/command/TestJMSInfoXCommand.java Wed Apr 17 23:55:20 2013
@@ -0,0 +1,93 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.oozie.command;
+
+import java.util.Properties;
+
+import org.apache.oozie.ErrorCode;
+import org.apache.oozie.WorkflowJobBean;
+import org.apache.oozie.client.WorkflowJob;
+import org.apache.oozie.client.rest.JMSConnectionInfoBean;
+import org.apache.oozie.jms.JMSJobEventListener;
+import org.apache.oozie.service.JMSAccessorService;
+import org.apache.oozie.service.JMSTopicService;
+import org.apache.oozie.service.Services;
+import org.apache.oozie.test.XDataTestCase;
+import org.apache.oozie.workflow.WorkflowInstance;
+import org.junit.Test;
+
+public class TestJMSInfoXCommand extends XDataTestCase {
+
+ private Services services;
+
+ @Override
+ protected void setUp() throws Exception {
+ super.setUp();
+ services = new Services();
+ services.init();
+ }
+
+ @Override
+ protected void tearDown() throws Exception {
+ services.destroy();
+ super.tearDown();
+ }
+
+ @Test
+ public void testConnectionNotEnabled() {
+ try {
+ WorkflowJobBean wfj = addRecordToWfJobTable(WorkflowJob.Status.SUCCEEDED, WorkflowInstance.Status.SUCCEEDED);
+ new JMSInfoXCommand(wfj.getId()).call();
+ }
+ catch (CommandException e) {
+ assertEquals(ErrorCode.E1601, e.getErrorCode());
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testJMSConnectionInfo() {
+ try {
+ services.destroy();
+ services = new Services();
+ services.getConf().set(
+ JMSJobEventListener.JMS_CONNECTION_PROPERTIES,
+ "java.naming.factory.initial#" + ActiveMQConnFactory + ";" + "java.naming.provider.url#"
+ + localActiveMQBroker + ";" + "connectionFactoryNames#" + "ConnectionFactory");
+ services.getConf().set(Services.CONF_SERVICE_EXT_CLASSES, JMSTopicService.class.getName());
+ services.init();
+
+ WorkflowJobBean wfj = addRecordToWfJobTable(WorkflowJob.Status.SUCCEEDED, WorkflowInstance.Status.SUCCEEDED);
+ JMSInfoXCommand jmsInfoCmd = new JMSInfoXCommand(wfj.getId());
+ JMSConnectionInfoBean jmsBean = jmsInfoCmd.call();
+ assertEquals("test", jmsBean.getTopicName());
+ Properties props = jmsBean.getJNDIProperties();
+ assertEquals(ActiveMQConnFactory, props.get("java.naming.factory.initial"));
+ assertEquals(localActiveMQBroker, props.get("java.naming.provider.url"));
+ assertEquals("ConnectionFactory", props.get("connectionFactoryNames"));
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+}
Modified: oozie/trunk/core/src/test/java/org/apache/oozie/event/TestEventQueue.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/test/java/org/apache/oozie/event/TestEventQueue.java?rev=1469108&r1=1469107&r2=1469108&view=diff
==============================================================================
--- oozie/trunk/core/src/test/java/org/apache/oozie/event/TestEventQueue.java (original)
+++ oozie/trunk/core/src/test/java/org/apache/oozie/event/TestEventQueue.java Wed Apr 17 23:55:20 2013
@@ -20,6 +20,8 @@ package org.apache.oozie.event;
import org.apache.hadoop.conf.Configuration;
import org.apache.oozie.client.WorkflowJob;
import org.apache.oozie.service.EventHandlerService;
+import org.apache.oozie.service.JMSAccessorService;
+import org.apache.oozie.service.JMSTopicService;
import org.apache.oozie.service.Services;
import org.apache.oozie.test.XDataTestCase;
@@ -28,16 +30,21 @@ import org.apache.oozie.test.XDataTestCa
*/
public class TestEventQueue extends XDataTestCase {
+ private Services services;
+
protected void setUp() throws Exception {
super.setUp();
- Services services = new Services();
+ services = new Services();
Configuration conf = services.getConf();
- conf.set(Services.CONF_SERVICE_EXT_CLASSES, "org.apache.oozie.service.EventHandlerService");
+ conf.set(Services.CONF_SERVICE_EXT_CLASSES,
+ JMSAccessorService.class.getName() + "," + JMSTopicService.class.getName() + ","
+ + EventHandlerService.class.getName());
+ conf.setInt(EventHandlerService.CONF_BATCH_SIZE, 3);
services.init();
}
protected void tearDown() throws Exception {
- Services.get().destroy();
+ services.destroy();
super.tearDown();
}
@@ -50,13 +57,7 @@ public class TestEventQueue extends XDat
}
public void testQueueOperations() throws Exception {
- Services services = Services.get();
- Configuration conf = services.getConf();
-
- // set smaller batch size for the events queue
- conf.setInt(EventHandlerService.CONF_BATCH_SIZE, 3);
EventHandlerService ehs = Services.get().get(EventHandlerService.class);
- ehs.init(services);
EventQueue eventQ = ehs.getEventQueue();
assertEquals(eventQ.getCurrentSize(), 0);
assertEquals(eventQ.getBatchSize(), 3);
Added: oozie/trunk/core/src/test/java/org/apache/oozie/jms/TestDefaultConnectionContext.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/test/java/org/apache/oozie/jms/TestDefaultConnectionContext.java?rev=1469108&view=auto
==============================================================================
--- oozie/trunk/core/src/test/java/org/apache/oozie/jms/TestDefaultConnectionContext.java (added)
+++ oozie/trunk/core/src/test/java/org/apache/oozie/jms/TestDefaultConnectionContext.java Wed Apr 17 23:55:20 2013
@@ -0,0 +1,107 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.oozie.jms;
+
+import javax.jms.Session;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.oozie.service.JMSAccessorService;
+import org.apache.oozie.service.JMSTopicService;
+import org.apache.oozie.service.Services;
+import org.apache.oozie.test.XTestCase;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class TestDefaultConnectionContext extends XTestCase {
+
+ private Services services;
+ private Session session1;
+ private Session session2;
+
+ @Before
+ protected void setUp() throws Exception {
+ super.setUp();
+ services = new Services();
+ Configuration conf = services.getConf();
+ conf.set(Services.CONF_SERVICE_EXT_CLASSES,
+ JMSAccessorService.class.getName() + "," + JMSTopicService.class.getName());
+ conf.set(JMSJobEventListener.JMS_CONNECTION_PROPERTIES, "java.naming.factory.initial#"
+ + ActiveMQConnFactory + ";" + "java.naming.provider.url#" + localActiveMQBroker
+ + ";connectionFactoryNames#" + "ConnectionFactory");
+ services.init();
+ }
+
+ @After
+ protected void tearDown() throws Exception {
+ services.destroy();
+ super.tearDown();
+ }
+
+ @Test
+ public void testThreadLocalSession() {
+ String jmsProps = services.getConf().get(JMSJobEventListener.JMS_CONNECTION_PROPERTIES);
+ JMSConnectionInfo connInfo = new JMSConnectionInfo(jmsProps);
+ ConnectionContext jmsContext = Services.get().get(JMSAccessorService.class)
+ .createConnectionContext(connInfo, false);
+ Thread th = new Thread(new SessionThread(jmsContext));
+ th.start();
+ try {
+ th.join();
+ }
+ catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ assertEquals(session1, session2);
+
+ ThreadLocal<Session> threadLocal1 = jmsContext.createThreadLocalSession(Session.AUTO_ACKNOWLEDGE);
+ Session session3 = threadLocal1.get();
+ ThreadLocal<Session> threadLocal2 = jmsContext.createThreadLocalSession(Session.AUTO_ACKNOWLEDGE);
+ Session session4 = threadLocal2.get();
+ // As session3 and session4 are created by same threads, they should be
+ // equal
+ assertTrue(session3.equals(session4));
+ // As session1 and session3 are created by diff threads, they shoudn't
+ // be equal
+ assertFalse(session1.equals(session3));
+ threadLocal1.remove();
+ ThreadLocal<Session> threadLocal3 = jmsContext.createThreadLocalSession(Session.AUTO_ACKNOWLEDGE);
+ Session session5 = threadLocal3.get();
+ assertFalse(session3.equals(session5));
+
+ }
+
+ class SessionThread implements Runnable {
+
+ private ConnectionContext connContext;
+
+ SessionThread(ConnectionContext connContext) {
+ this.connContext = connContext;
+ }
+
+ @Override
+ public void run() {
+ ThreadLocal<Session> th1 = connContext.createThreadLocalSession(Session.AUTO_ACKNOWLEDGE);
+ session1 = th1.get();
+ ThreadLocal<Session> th2 = connContext.createThreadLocalSession(Session.AUTO_ACKNOWLEDGE);
+ session2 = th2.get();
+ }
+
+ }
+
+}