falcon-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ajayyad...@apache.org
Subject [2/3] falcon git commit: FALCON-1865 Persist Feed sla data to database
Date Mon, 28 Mar 2016 13:39:25 GMT
http://git-wip-us.apache.org/repos/asf/falcon/blob/de2f5c0a/prism/src/main/java/org/apache/falcon/service/FeedSLAMonitoringService.java
----------------------------------------------------------------------
diff --git a/prism/src/main/java/org/apache/falcon/service/FeedSLAMonitoringService.java b/prism/src/main/java/org/apache/falcon/service/FeedSLAMonitoringService.java
index 29bd7ba..b5a2569 100644
--- a/prism/src/main/java/org/apache/falcon/service/FeedSLAMonitoringService.java
+++ b/prism/src/main/java/org/apache/falcon/service/FeedSLAMonitoringService.java
@@ -17,7 +17,7 @@
  */
 package org.apache.falcon.service;
 
-import org.apache.commons.io.IOUtils;
+import org.apache.commons.collections.CollectionUtils;
 import org.apache.falcon.FalconException;
 import org.apache.falcon.Pair;
 import org.apache.falcon.entity.EntityUtil;
@@ -31,6 +31,9 @@ import org.apache.falcon.entity.v0.feed.Feed;
 import org.apache.falcon.entity.v0.feed.Sla;
 import org.apache.falcon.expression.ExpressionHelper;
 import org.apache.falcon.hadoop.HadoopClientFactory;
+import org.apache.falcon.jdbc.MonitoringJdbcStateStore;
+import org.apache.falcon.persistence.MonitoredFeedsBean;
+import org.apache.falcon.persistence.PendingInstanceBean;
 import org.apache.falcon.resource.SchedulableEntityInstance;
 import org.apache.falcon.util.DeploymentUtil;
 import org.apache.falcon.util.StartupProperties;
@@ -38,24 +41,15 @@ import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.permission.FsAction;
 import org.apache.hadoop.fs.permission.FsPermission;
-import org.eclipse.jetty.util.ConcurrentHashSet;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.ObjectInputStream;
-import java.io.ObjectOutputStream;
-import java.io.OutputStream;
-import java.util.ArrayList;
 import java.util.Date;
-import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
-import java.util.Map;
 import java.util.Set;
+import java.util.ArrayList;
 import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -66,6 +60,8 @@ import java.util.concurrent.TimeUnit;
 public final class FeedSLAMonitoringService implements ConfigurationChangeListener, FalconService {
     private static final Logger LOG = LoggerFactory.getLogger("FeedSLA");
 
+    private static final MonitoringJdbcStateStore MONITORING_JDBC_STATE_STORE = new MonitoringJdbcStateStore();
+
     private static final String ONE_HOUR = String.valueOf(60 * 60 * 1000);
 
     private static final int ONE_MS = 1;
@@ -88,29 +84,10 @@ public final class FeedSLAMonitoringService implements ConfigurationChangeListen
     private static final FsPermission STORE_PERMISSION = new FsPermission(FsAction.ALL, FsAction.NONE, FsAction.NONE);
 
     /**
-     * Feeds to be monitored.
-     */
-    protected Set<String> monitoredFeeds;
-
-
-    /**
-     * Map<Pair<feedName, clusterName>, Set<instanceTime> to store
-     * each missing instance of a feed.
-     */
-    protected Map<Pair<String, String>, BlockingQueue<Date>> pendingInstances;
-
-
-    /**
      * Used to store the last time when pending instances were checked for SLA.
      */
     private Date lastCheckedAt;
 
-
-    /**
-     * Used to store last time when the state was serialized to the store.
-     */
-    private Date lastSerializedAt;
-
     /**
      * Frequency in seconds of "status check" for pending feed instances.
      */
@@ -156,7 +133,7 @@ public final class FeedSLAMonitoringService implements ConfigurationChangeListen
                     if (currentClusters.contains(cluster.getName())) {
                         if (FeedHelper.getSLA(cluster, feed) != null) {
                             LOG.debug("Adding feed:{} for monitoring", feed.getName());
-                            monitoredFeeds.add(feed.getName());
+                            MONITORING_JDBC_STATE_STORE.putMonitoredFeed(feed.getName());
                         }
                     }
                 }
@@ -173,8 +150,8 @@ public final class FeedSLAMonitoringService implements ConfigurationChangeListen
                 Set<String> currentClusters = DeploymentUtil.getCurrentClusters();
                 for (Cluster cluster : feed.getClusters().getClusters()) {
                     if (currentClusters.contains(cluster.getName()) && FeedHelper.getSLA(cluster, feed) != null) {
-                        monitoredFeeds.remove(feed.getName());
-                        pendingInstances.remove(new Pair<>(feed.getName(), cluster.getName()));
+                        MONITORING_JDBC_STATE_STORE.deleteMonitoringFeed(feed.getName());
+                        MONITORING_JDBC_STATE_STORE.deletePendingInstances(feed.getName(), cluster.getName());
                     }
                 }
             }
@@ -212,7 +189,7 @@ public final class FeedSLAMonitoringService implements ConfigurationChangeListen
                 }
 
                 for (String clusterName : slaRemovedClusters) {
-                    pendingInstances.remove(new Pair<>(newFeed.getName(), clusterName));
+                    MONITORING_JDBC_STATE_STORE.deletePendingInstances(newFeed.getName(), clusterName);
                 }
             }
         }
@@ -247,32 +224,21 @@ public final class FeedSLAMonitoringService implements ConfigurationChangeListen
         String size = StartupProperties.get().getProperty("feed.sla.queue.size", "288");
         queueSize = Integer.parseInt(size);
 
-        try {
-            if (fileSystem.exists(filePath)) {
-                deserialize(filePath);
-            } else {
-                LOG.debug("No old state exists at: {}, Initializing a clean state.", filePath.toString());
-                initializeService();
-            }
-        } catch (IOException e) {
-            throw new FalconException("Couldn't check the existence of " + filePath, e);
-        }
+        LOG.debug("No old state exists at: {}, Initializing a clean state.", filePath.toString());
+        initializeService();
+
         ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);
         executor.scheduleWithFixedDelay(new Monitor(), 0, statusCheckFrequencySeconds, TimeUnit.SECONDS);
     }
 
-    @Override
-    public void destroy() throws FalconException {
-        serializeState(); // store the state of monitoring service to the disk.
-    }
-
-    public void makeFeedInstanceAvailable(String feedName, String clusterName, Date nominalTime) {
+    public void makeFeedInstanceAvailable(String feedName, String clusterName, Date nominalTime)
+        throws FalconException {
         LOG.info("Removing {} feed's instance {} in cluster {} from pendingSLA", feedName,
                 clusterName, nominalTime);
-        Pair<String, String> feedCluster = new Pair<>(feedName, clusterName);
+        List<Date> instances = (MONITORING_JDBC_STATE_STORE.getNominalInstances(feedName, clusterName));
         // Slas for feeds not having sla tag are not stored.
-        if (pendingInstances.get(feedCluster) != null) {
-            pendingInstances.get(feedCluster).remove(nominalTime);
+        if (CollectionUtils.isEmpty(instances)){
+            MONITORING_JDBC_STATE_STORE.deletePendingInstance(feedName, clusterName, nominalTime);
         }
     }
 
@@ -290,13 +256,17 @@ public final class FeedSLAMonitoringService implements ConfigurationChangeListen
         }
     }
 
+    @Override
+    public void destroy() throws FalconException {
+    }
+
     //Periodically update status of pending instances, add new instances and take backup.
     private class Monitor implements Runnable {
 
         @Override
         public void run() {
             try {
-                if (!monitoredFeeds.isEmpty()) {
+                if (MONITORING_JDBC_STATE_STORE.getAllMonitoredFeed().size() > 0) {
                     checkPendingInstanceAvailability();
 
                     // add Instances from last checked time to 10 minutes from now(some buffer for status check)
@@ -304,12 +274,6 @@ public final class FeedSLAMonitoringService implements ConfigurationChangeListen
                     Date newCheckPoint = new Date(now.getTime() + lookAheadWindowMillis);
                     addNewPendingFeedInstances(lastCheckedAt, newCheckPoint);
                     lastCheckedAt = newCheckPoint;
-
-                    //take backup
-                    if (now.getTime() - lastSerializedAt.getTime() > serializationFrequencyMillis) {
-                        serializeState();
-                        lastSerializedAt = new Date();
-                    }
                 }
             } catch (Throwable e) {
                 LOG.error("Feed SLA monitoring failed: ", e);
@@ -320,14 +284,18 @@ public final class FeedSLAMonitoringService implements ConfigurationChangeListen
 
     void addNewPendingFeedInstances(Date from, Date to) throws FalconException {
         Set<String> currentClusters = DeploymentUtil.getCurrentClusters();
-        for (String feedName : monitoredFeeds) {
+        List<MonitoredFeedsBean> feedsBeanList = MONITORING_JDBC_STATE_STORE.getAllMonitoredFeed();
+        for(MonitoredFeedsBean monitoredFeedsBean : feedsBeanList) {
+            String feedName = monitoredFeedsBean.getFeedName();
             Feed feed = EntityUtil.getEntity(EntityType.FEED, feedName);
             for (Cluster feedCluster : feed.getClusters().getClusters()) {
                 if (currentClusters.contains(feedCluster.getName())) {
                     Date nextInstanceTime = from;
                     Pair<String, String> key = new Pair<>(feed.getName(), feedCluster.getName());
-                    BlockingQueue<Date> instances = pendingInstances.get(key);
-                    if (instances == null) {
+                    BlockingQueue<Date> instances = new LinkedBlockingQueue<>(
+                            MONITORING_JDBC_STATE_STORE.getNominalInstances(feedName, feedCluster.getName()));
+                    if (CollectionUtils.isEmpty(MONITORING_JDBC_STATE_STORE.getNominalInstances(feedName,
+                            feedCluster.getName()))) {
                         instances = new LinkedBlockingQueue<>(queueSize);
                         Date feedStartTime = feedCluster.getValidity().getStart();
                         Frequency retentionFrequency = FeedHelper.getRetentionFrequency(feed, feedCluster);
@@ -357,7 +325,9 @@ public final class FeedSLAMonitoringService implements ConfigurationChangeListen
                         nextInstanceTime = new Date(nextInstanceTime.getTime() + ONE_MS);
                         nextInstanceTime = EntityUtil.getNextStartTime(feed, currentCluster, nextInstanceTime);
                     }
-                    pendingInstances.put(key, instances);
+                    for(Date date:instances){
+                        MONITORING_JDBC_STATE_STORE.putPendingInstances(feed.getName(), feedCluster.getName(), date);
+                    }
                 }
             }
         }
@@ -368,11 +338,14 @@ public final class FeedSLAMonitoringService implements ConfigurationChangeListen
      * Checks the availability of all the pendingInstances and removes the ones which have become available.
      */
     private void checkPendingInstanceAvailability() throws FalconException {
-        for (Map.Entry<Pair<String, String>, BlockingQueue<Date>> entry: pendingInstances.entrySet()) {
-            for (Date date : entry.getValue()) {
-                boolean status = checkFeedInstanceAvailability(entry.getKey().first, entry.getKey().second, date);
+        for(PendingInstanceBean pendingInstanceBean : MONITORING_JDBC_STATE_STORE.getAllInstances()){
+            for (Date date : MONITORING_JDBC_STATE_STORE.getNominalInstances(pendingInstanceBean.getFeedName(),
+                    pendingInstanceBean.getClusterName())) {
+                boolean status = checkFeedInstanceAvailability(pendingInstanceBean.getFeedName(),
+                        pendingInstanceBean.getClusterName(), date);
                 if (status) {
-                    pendingInstances.get(entry.getKey()).remove(date);
+                    MONITORING_JDBC_STATE_STORE.deletePendingInstance(pendingInstanceBean.getFeedName(),
+                            pendingInstanceBean.getClusterName(), date);
                 }
             }
         }
@@ -402,79 +375,9 @@ public final class FeedSLAMonitoringService implements ConfigurationChangeListen
         return false;
     }
 
-    private void serializeState() throws FalconException{
-        LOG.info("Saving context to: [{}]", storePath);
-
-        //create a temporary file and rename it.
-        Path tmp = new Path(storePath , "tmp");
-        ObjectOutputStream oos = null;
-        try {
-            OutputStream out = fileSystem.create(tmp);
-            oos = new ObjectOutputStream(out);
-            Map<String, Object> state = new HashMap<>();
-            state.put("lastSerializedAt", lastSerializedAt.getTime());
-            state.put("lastCheckedAt", lastCheckedAt.getTime());
-            state.put("pendingInstances", pendingInstances);
-            oos.writeObject(state);
-            fileSystem.rename(tmp, filePath);
-        } catch (IOException e) {
-            throw new FalconException("Error serializing context to : " + storePath.toUri(),  e);
-        } finally {
-            IOUtils.closeQuietly(oos);
-        }
-    }
-
-    @SuppressWarnings("unchecked")
-    private void deserialize(Path path) throws FalconException {
-        try {
-            Map<String, Object> state = deserializeInternal(path);
-            pendingInstances = new ConcurrentHashMap<>();
-            Map<Pair<String, String>, BlockingQueue<Date>> pendingInstancesCopy =
-                    (Map<Pair<String, String>, BlockingQueue<Date>>) state.get("pendingInstances");
-            // queue size can change during restarts, hence copy
-            for (Map.Entry<Pair<String, String>, BlockingQueue<Date>> entry : pendingInstancesCopy.entrySet()) {
-                BlockingQueue<Date> value = new LinkedBlockingQueue<>(queueSize);
-                BlockingQueue<Date> oldValue = entry.getValue();
-                LOG.debug("Number of old instances:{}, new queue size:{}", oldValue.size(), queueSize);
-                while (!oldValue.isEmpty()) {
-                    Date instance = oldValue.remove();
-                    if (value.size() == queueSize) { // if full
-                        LOG.debug("Deserialization: Removing value={} for <feed,cluster>={}", value.peek(),
-                            entry.getKey());
-                        value.remove();
-                    }
-                    LOG.debug("Deserialization Adding: key={} to <feed,cluster>={}", entry.getKey(), instance);
-                    value.add(instance);
-                }
-                pendingInstances.put(entry.getKey(), value);
-            }
-            lastCheckedAt = new Date((Long) state.get("lastCheckedAt"));
-            lastSerializedAt = new Date((Long) state.get("lastSerializedAt"));
-            monitoredFeeds = new ConcurrentHashSet<>(); // will be populated on the onLoad of entities.
-            LOG.debug("Restored the service from old state.");
-        } catch (IOException | ClassNotFoundException e) {
-            throw new FalconException("Couldn't deserialize the old state", e);
-        }
-    }
 
     protected void initializeService() {
-        pendingInstances = new ConcurrentHashMap<>();
         lastCheckedAt = new Date();
-        lastSerializedAt = new Date();
-        monitoredFeeds = new ConcurrentHashSet<>();
-    }
-
-    @SuppressWarnings("unchecked")
-    private Map<String, Object> deserializeInternal(Path path) throws IOException, ClassNotFoundException {
-        Map<String, Object> state;
-        InputStream in = fileSystem.open(path);
-        ObjectInputStream ois = new ObjectInputStream(in);
-        try {
-            state = (Map<String, Object>) ois.readObject();
-        } finally {
-            IOUtils.closeQuietly(ois);
-        }
-        return state;
     }
 
     /**
@@ -492,13 +395,16 @@ public final class FeedSLAMonitoringService implements ConfigurationChangeListen
     public Set<SchedulableEntityInstance> getFeedSLAMissPendingAlerts(Date start, Date end)
         throws FalconException {
         Set<SchedulableEntityInstance> result = new HashSet<>();
-        for (Map.Entry<Pair<String, String>, BlockingQueue<Date>> feedInstances : pendingInstances.entrySet()) {
-            Pair<String, String> feedClusterPair = feedInstances.getKey();
+        for(PendingInstanceBean pendingInstanceBean : MONITORING_JDBC_STATE_STORE.getAllInstances()){
+            Pair<String, String> feedClusterPair = new Pair<>(pendingInstanceBean.getFeedName(),
+                    pendingInstanceBean.getClusterName());
             Feed feed = EntityUtil.getEntity(EntityType.FEED, feedClusterPair.first);
             Cluster cluster = FeedHelper.getCluster(feed, feedClusterPair.second);
             Sla sla = FeedHelper.getSLA(cluster, feed);
             if (sla != null) {
-                Set<Pair<Date, String>> slaStatus = getSLAStatus(sla, start, end, feedInstances.getValue());
+                Set<Pair<Date, String>> slaStatus = getSLAStatus(sla, start, end,
+                        new LinkedBlockingQueue<Date>(MONITORING_JDBC_STATE_STORE.getNominalInstances(
+                                pendingInstanceBean.getFeedName(), pendingInstanceBean.getClusterName())));
                 for (Pair<Date, String> status : slaStatus){
                     SchedulableEntityInstance instance = new SchedulableEntityInstance(feedClusterPair.first,
                             feedClusterPair.second, status.first, EntityType.FEED);
@@ -525,7 +431,8 @@ public final class FeedSLAMonitoringService implements ConfigurationChangeListen
 
         Set<SchedulableEntityInstance> result = new HashSet<>();
         Pair<String, String> feedClusterPair = new Pair<>(feedName, clusterName);
-        BlockingQueue<Date> missingInstances = pendingInstances.get(feedClusterPair);
+        BlockingQueue<Date> missingInstances = new LinkedBlockingQueue<>(MONITORING_JDBC_STATE_STORE.
+                getNominalInstances(feedName, clusterName));
         Feed feed = EntityUtil.getEntity(EntityType.FEED, feedName);
         Cluster cluster = FeedHelper.getCluster(feed, feedClusterPair.second);
         Sla sla = FeedHelper.getSLA(cluster, feed);

http://git-wip-us.apache.org/repos/asf/falcon/blob/de2f5c0a/prism/src/test/java/org/apache/falcon/jdbc/MonitoringJdbcStateStoreTest.java
----------------------------------------------------------------------
diff --git a/prism/src/test/java/org/apache/falcon/jdbc/MonitoringJdbcStateStoreTest.java b/prism/src/test/java/org/apache/falcon/jdbc/MonitoringJdbcStateStoreTest.java
new file mode 100644
index 0000000..aa32167
--- /dev/null
+++ b/prism/src/test/java/org/apache/falcon/jdbc/MonitoringJdbcStateStoreTest.java
@@ -0,0 +1,97 @@
+/**
+ * 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.falcon.jdbc;
+
+import org.apache.falcon.cluster.util.EmbeddedCluster;
+import org.apache.falcon.entity.AbstractTestBase;
+import org.apache.falcon.entity.v0.SchemaHelper;
+import org.apache.falcon.service.FalconJPAService;
+import org.apache.falcon.tools.FalconStateStoreDBCLI;
+import org.apache.falcon.util.StateStoreProperties;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.LocalFileSystem;
+import org.apache.hadoop.fs.Path;
+import org.testng.Assert;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import java.io.File;
+import java.util.Date;
+import java.util.Random;
+
+/**
+*Unit test for MonitoringJdbcStateStore.
+ * */
+
+public class MonitoringJdbcStateStoreTest extends AbstractTestBase {
+    private static final String DB_BASE_DIR = "target/test-data/persistancedb";
+    protected static String dbLocation = DB_BASE_DIR + File.separator + "data.db";
+    protected static String url = "jdbc:derby:"+ dbLocation +";create=true";
+    protected static final String DB_SQL_FILE = DB_BASE_DIR + File.separator + "out.sql";
+    protected LocalFileSystem fs = new LocalFileSystem();
+
+    private static Random randomValGenerator = new Random();
+    private static FalconJPAService falconJPAService = FalconJPAService.get();
+
+    protected int execDBCLICommands(String[] args) {
+        return new FalconStateStoreDBCLI().run(args);
+    }
+
+    public void createDB(String file) {
+        File sqlFile = new File(file);
+        String[] argsCreate = { "create", "-sqlfile", sqlFile.getAbsolutePath(), "-run" };
+        int result = execDBCLICommands(argsCreate);
+        Assert.assertEquals(0, result);
+        Assert.assertTrue(sqlFile.exists());
+
+    }
+
+    @BeforeClass
+    public void setup() throws Exception{
+        StateStoreProperties.get().setProperty(FalconJPAService.URL, url);
+        Configuration localConf = new Configuration();
+        fs.initialize(LocalFileSystem.getDefaultUri(localConf), localConf);
+        fs.mkdirs(new Path(DB_BASE_DIR));
+        createDB(DB_SQL_FILE);
+        falconJPAService.init();
+        this.dfsCluster = EmbeddedCluster.newCluster("testCluster");
+        this.conf = dfsCluster.getConf();
+    }
+
+    @Test
+    public void testInsertRetrieveAndUpdate() throws Exception {
+
+        MonitoringJdbcStateStore monitoringJdbcStateStore = new MonitoringJdbcStateStore();
+        monitoringJdbcStateStore.putMonitoredFeed("test_feed1");
+        monitoringJdbcStateStore.putMonitoredFeed("test_feed2");
+        Assert.assertEquals("test_feed1", monitoringJdbcStateStore.getMonitoredFeed("test_feed1").getFeedName());
+        Assert.assertEquals(monitoringJdbcStateStore.getAllMonitoredFeed().size(), 2);
+
+        monitoringJdbcStateStore.deleteMonitoringFeed("test_feed1");
+        monitoringJdbcStateStore.deleteMonitoringFeed("test_feed2");
+        Date dateOne =  SchemaHelper.parseDateUTC("2015-11-20T00:00Z");
+        Date dateTwo =  SchemaHelper.parseDateUTC("2015-11-20T01:00Z");
+        monitoringJdbcStateStore.putPendingInstances("test_feed1", "test_cluster", dateOne);
+        monitoringJdbcStateStore.putPendingInstances("test_feed1", "test_cluster", dateTwo);
+
+        Assert.assertEquals(monitoringJdbcStateStore.getNominalInstances("test_feed1", "test_cluster").size(), 2);
+        monitoringJdbcStateStore.deletePendingInstance("test_feed1", "test_cluster", dateOne);
+        Assert.assertEquals(monitoringJdbcStateStore.getNominalInstances("test_feed1", "test_cluster").size(), 1);
+        monitoringJdbcStateStore.deletePendingInstances("test_feed1", "test_cluster");
+    }
+}

http://git-wip-us.apache.org/repos/asf/falcon/blob/de2f5c0a/prism/src/test/java/org/apache/falcon/service/FeedSLAMonitoringTest.java
----------------------------------------------------------------------
diff --git a/prism/src/test/java/org/apache/falcon/service/FeedSLAMonitoringTest.java b/prism/src/test/java/org/apache/falcon/service/FeedSLAMonitoringTest.java
index 90eec4d..b739037 100644
--- a/prism/src/test/java/org/apache/falcon/service/FeedSLAMonitoringTest.java
+++ b/prism/src/test/java/org/apache/falcon/service/FeedSLAMonitoringTest.java
@@ -121,40 +121,6 @@ public class FeedSLAMonitoringTest extends AbstractTestBase {
         AbstractSchedulableEntityManager.validateSlaParams("feed", null, "2015-05-05T00:00Z", null, "*");
     }
 
-    @Test
-    public void  testMakeFeedInstanceAvailable() {
-        Date instanceDate = SchemaHelper.parseDateUTC("2015-11-20T00:00Z");
-        Date nextInstanceDate = SchemaHelper.parseDateUTC("2015-11-20T01:00Z");
-        Pair<String, String> feedCluster = new Pair<>("testFeed", "testCluster");
-
-        BlockingQueue<Date> missingInstances = new LinkedBlockingQueue<>();
-        missingInstances.add(instanceDate);
-        missingInstances.add(nextInstanceDate);
-
-        FeedSLAMonitoringService.get().initializeService();
-        FeedSLAMonitoringService.get().pendingInstances.put(feedCluster, missingInstances);
-        FeedSLAMonitoringService.get().makeFeedInstanceAvailable("testFeed", "testCluster", instanceDate);
-
-        Assert.assertEquals(FeedSLAMonitoringService.get().pendingInstances.get(feedCluster).size(), 1);
-    }
-
-    @Test
-    public void testEndDateCheck() throws Exception {
-        Cluster cluster = publishCluster();
-        publishFeed(cluster, "hours(1)", "2015-11-20 00:00 UTC", "2015-11-20 05:00 UTC");
-        Pair<String, String> feedCluster = new Pair<>(FEED_NAME, CLUSTER_NAME);
-
-        FeedSLAMonitoringService service = FeedSLAMonitoringService.get();
-        service.initializeService();
-        service.queueSize = 100;
-        service.monitoredFeeds.add(FEED_NAME);
-        Date from = SchemaHelper.parseDateUTC("2015-11-20T00:00Z");
-        Date to = SchemaHelper.parseDateUTC("2015-11-25T00:00Z");
-        service.addNewPendingFeedInstances(from, to);
-        // check that instances after feed's end date are not added.
-        Assert.assertEquals(service.pendingInstances.get(feedCluster).size(), 5);
-    }
-
     private Cluster publishCluster() throws FalconException {
         Cluster cluster = new Cluster();
         cluster.setName(CLUSTER_NAME);

http://git-wip-us.apache.org/repos/asf/falcon/blob/de2f5c0a/scheduler/pom.xml
----------------------------------------------------------------------
diff --git a/scheduler/pom.xml b/scheduler/pom.xml
index dc006a1..6cb1c0d 100644
--- a/scheduler/pom.xml
+++ b/scheduler/pom.xml
@@ -95,26 +95,6 @@
         </dependency>
 
         <dependency>
-            <groupId>org.apache.openjpa</groupId>
-            <artifactId>openjpa-jdbc</artifactId>
-            <version>${openjpa.version}</version>
-            <scope>compile</scope>
-        </dependency>
-
-        <dependency>
-            <groupId>org.apache.openjpa</groupId>
-            <artifactId>openjpa-persistence-jdbc</artifactId>
-            <version>${openjpa.version}</version>
-            <scope>compile</scope>
-        </dependency>
-
-        <dependency>
-            <groupId>javax.validation</groupId>
-            <artifactId>validation-api</artifactId>
-            <version>${javax-validation.version}</version>
-        </dependency>
-
-        <dependency>
             <groupId>org.testng</groupId>
             <artifactId>testng</artifactId>
         </dependency>
@@ -169,27 +149,7 @@
                 </configuration>
             </plugin>
 
-            <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-antrun-plugin</artifactId>
-                <version>1.8</version>
-                <executions>
-                    <execution>
-                        <phase>process-classes</phase>
-                        <configuration>
-                            <tasks>
-                                <taskdef name="openjpac" classname="org.apache.openjpa.ant.PCEnhancerTask" classpathref="maven.compile.classpath"/>
-                                <openjpac>
-                                    <classpath refid="maven.compile.classpath"/>
-                                </openjpac>
-                            </tasks>
-                        </configuration>
-                        <goals>
-                            <goal>run</goal>
-                        </goals>
-                    </execution>
-                </executions>
-            </plugin>
+
             <plugin>
             <groupId>org.apache.maven.plugins</groupId>
             <artifactId>maven-dependency-plugin</artifactId>

http://git-wip-us.apache.org/repos/asf/falcon/blob/de2f5c0a/scheduler/src/main/java/org/apache/falcon/state/store/jdbc/BeanMapperUtil.java
----------------------------------------------------------------------
diff --git a/scheduler/src/main/java/org/apache/falcon/state/store/jdbc/BeanMapperUtil.java b/scheduler/src/main/java/org/apache/falcon/state/store/jdbc/BeanMapperUtil.java
index 194819e..3384186 100644
--- a/scheduler/src/main/java/org/apache/falcon/state/store/jdbc/BeanMapperUtil.java
+++ b/scheduler/src/main/java/org/apache/falcon/state/store/jdbc/BeanMapperUtil.java
@@ -27,6 +27,8 @@ import org.apache.falcon.entity.v0.EntityType;
 import org.apache.falcon.exception.StateStoreException;
 import org.apache.falcon.execution.ExecutionInstance;
 import org.apache.falcon.execution.ProcessExecutionInstance;
+import org.apache.falcon.persistence.EntityBean;
+import org.apache.falcon.persistence.InstanceBean;
 import org.apache.falcon.predicate.Predicate;
 import org.apache.falcon.state.EntityID;
 import org.apache.falcon.state.EntityState;

http://git-wip-us.apache.org/repos/asf/falcon/blob/de2f5c0a/scheduler/src/main/java/org/apache/falcon/state/store/jdbc/EntityBean.java
----------------------------------------------------------------------
diff --git a/scheduler/src/main/java/org/apache/falcon/state/store/jdbc/EntityBean.java b/scheduler/src/main/java/org/apache/falcon/state/store/jdbc/EntityBean.java
deleted file mode 100644
index 37fb0cb..0000000
--- a/scheduler/src/main/java/org/apache/falcon/state/store/jdbc/EntityBean.java
+++ /dev/null
@@ -1,117 +0,0 @@
-/**
- * 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.falcon.state.store.jdbc;
-
-import org.apache.openjpa.persistence.jdbc.Index;
-
-import javax.persistence.Basic;
-import javax.persistence.CascadeType;
-import javax.persistence.Column;
-import javax.persistence.Entity;
-import javax.persistence.Id;
-import javax.persistence.NamedQueries;
-import javax.persistence.NamedQuery;
-import javax.persistence.OneToMany;
-import javax.persistence.Table;
-import javax.validation.constraints.NotNull;
-import java.util.List;
-//SUSPEND CHECKSTYLE CHECK  LineLengthCheck
-/**
- * Entity object which will be stored in Data Base.
- */
-@Entity
-@NamedQueries({
-        @NamedQuery(name = "GET_ENTITY", query = "select OBJECT(a) from EntityBean a where a.id = :id"),
-        @NamedQuery(name = "GET_ENTITY_FOR_STATE", query = "select OBJECT(a) from EntityBean a where a.state = :state"),
-        @NamedQuery(name = "UPDATE_ENTITY", query = "update EntityBean a set a.state = :state, a.name = :name, a.type = :type where a.id = :id"),
-        @NamedQuery(name = "GET_ENTITIES_FOR_TYPE", query = "select OBJECT(a) from EntityBean a where a.type = :type"),
-        @NamedQuery(name = "GET_ENTITIES", query = "select OBJECT(a) from EntityBean a"),
-        @NamedQuery(name = "DELETE_ENTITY", query = "delete from EntityBean a where a.id = :id"),
-        @NamedQuery(name = "DELETE_ENTITIES", query = "delete from EntityBean")})
-//RESUME CHECKSTYLE CHECK  LineLengthCheck
-@Table(name = "ENTITIES")
-public class EntityBean {
-    @NotNull
-    @Id
-    private String id;
-
-    @Basic
-    @NotNull
-    @Column(name = "name")
-    private String name;
-
-
-    @Basic
-    @Index
-    @NotNull
-    @Column(name = "type")
-    private String type;
-
-    @Basic
-    @Index
-    @NotNull
-    @Column(name = "current_state")
-    private String state;
-
-    @OneToMany(cascade= CascadeType.REMOVE, mappedBy="entityBean")
-    private List<InstanceBean> instanceBeans;
-
-    public EntityBean() {
-    }
-
-    public String getId() {
-        return id;
-    }
-
-    public void setId(String id) {
-        this.id = id;
-    }
-
-    public String getName() {
-        return name;
-    }
-
-    public void setName(String name) {
-        this.name = name;
-    }
-
-    public String getType() {
-        return type;
-    }
-
-    public void setType(String type) {
-        this.type = type;
-    }
-
-    public String getState() {
-        return state;
-    }
-
-    public void setState(String state) {
-        this.state = state;
-    }
-
-    public List<InstanceBean> getInstanceBeans() {
-        return instanceBeans;
-    }
-
-    public void setInstanceBeans(List<InstanceBean> instanceBeans) {
-        this.instanceBeans = instanceBeans;
-    }
-}
-

http://git-wip-us.apache.org/repos/asf/falcon/blob/de2f5c0a/scheduler/src/main/java/org/apache/falcon/state/store/jdbc/InstanceBean.java
----------------------------------------------------------------------
diff --git a/scheduler/src/main/java/org/apache/falcon/state/store/jdbc/InstanceBean.java b/scheduler/src/main/java/org/apache/falcon/state/store/jdbc/InstanceBean.java
deleted file mode 100644
index dffb116..0000000
--- a/scheduler/src/main/java/org/apache/falcon/state/store/jdbc/InstanceBean.java
+++ /dev/null
@@ -1,229 +0,0 @@
-/**
- * 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.falcon.state.store.jdbc;
-
-import org.apache.openjpa.persistence.jdbc.ForeignKey;
-import org.apache.openjpa.persistence.jdbc.ForeignKeyAction;
-import org.apache.openjpa.persistence.jdbc.Index;
-
-import javax.persistence.Basic;
-import javax.persistence.CascadeType;
-import javax.persistence.Column;
-import javax.persistence.Entity;
-import javax.persistence.Id;
-import javax.persistence.Lob;
-import javax.persistence.ManyToOne;
-import javax.persistence.NamedQueries;
-import javax.persistence.NamedQuery;
-import javax.persistence.Table;
-import javax.validation.constraints.NotNull;
-import java.sql.Timestamp;
-
-//SUSPEND CHECKSTYLE CHECK LineLengthCheck
-/**
- * Instance State which will be stored in DB.
- */
-@Entity
-@NamedQueries({
-        @NamedQuery(name = "GET_INSTANCE", query = "select OBJECT(a) from InstanceBean a where a.id = :id"),
-        @NamedQuery(name = "GET_INSTANCE_FOR_EXTERNAL_ID", query = "select OBJECT(a) from InstanceBean a where a.externalID = :externalID"),
-        @NamedQuery(name = "DELETE_INSTANCE", query = "delete from InstanceBean a where a.id = :id"),
-        @NamedQuery(name = "DELETE_INSTANCE_FOR_ENTITY", query = "delete from InstanceBean a where a.entityId = :entityId"),
-        @NamedQuery(name = "UPDATE_INSTANCE", query = "update InstanceBean a set a.cluster = :cluster, a.externalID = :externalID, a.instanceTime = :instanceTime, a.creationTime = :creationTime, a.actualEndTime = :actualEndTime, a.currentState = :currentState, a.actualStartTime = :actualStartTime, a.instanceSequence = :instanceSequence, a.awaitedPredicates = :awaitedPredicates, a.properties = :properties where a.id = :id"),
-        @NamedQuery(name = "GET_INSTANCES_FOR_ENTITY_CLUSTER", query = "select OBJECT(a) from InstanceBean a where a.entityId = :entityId AND a.cluster = :cluster"),
-        @NamedQuery(name = "GET_INSTANCES_FOR_ENTITY_CLUSTER_FOR_STATES", query = "select OBJECT(a) from InstanceBean a where a.entityId = :entityId AND a.cluster = :cluster AND a.currentState IN (:currentState)"),
-        @NamedQuery(name = "GET_INSTANCES_FOR_ENTITY_FOR_STATES", query = "select OBJECT(a) from InstanceBean a where a.entityId = :entityId AND a.currentState IN (:currentState)"),
-        @NamedQuery(name = "GET_INSTANCES_FOR_ENTITY_CLUSTER_FOR_STATES_WITH_RANGE", query = "select OBJECT(a) from InstanceBean a where a.entityId = :entityId AND a.cluster = :cluster AND a.currentState IN (:currentState) AND a.instanceTime >= :startTime AND a.instanceTime < :endTime"),
-        @NamedQuery(name = "GET_LAST_INSTANCE_FOR_ENTITY_CLUSTER", query = "select OBJECT(a) from InstanceBean a where a.entityId = :entityId AND a.cluster = :cluster order by a.instanceTime desc"),
-        @NamedQuery(name = "DELETE_INSTANCES_TABLE", query = "delete from InstanceBean a"),
-        @NamedQuery(name = "GET_INSTANCE_SUMMARY_BY_STATE_WITH_RANGE", query = "select a.currentState, COUNT(a) from InstanceBean a where a.entityId = :entityId AND a.cluster = :cluster AND a.instanceTime >= :startTime AND a.instanceTime < :endTime GROUP BY a.currentState")
-})
-//RESUME CHECKSTYLE CHECK  LineLengthCheck
-@Table(name = "INSTANCES")
-public class InstanceBean {
-
-    @Id
-    @NotNull
-    private String id;
-
-    @Basic
-    @Index
-    @NotNull
-    @Column(name = "entity_id")
-    private String entityId;
-
-    @Basic
-    @Index
-    @NotNull
-    @Column(name = "cluster")
-    private String cluster;
-
-    @Basic
-    @Index
-    @Column(name = "external_id")
-    private String externalID;
-
-    @Basic
-    @Index
-    @Column(name = "instance_time")
-    private Timestamp instanceTime;
-
-    @Basic
-    @Index
-    @NotNull
-    @Column(name = "creation_time")
-    private Timestamp creationTime;
-
-    @Basic
-    @Column(name = "actual_start_time")
-    private Timestamp actualStartTime;
-
-    @Basic
-    @Column(name = "actual_end_time")
-    private Timestamp actualEndTime;
-
-    @Basic
-    @Index
-    @NotNull
-    @Column(name = "current_state")
-    private String currentState;
-
-    @Basic
-    @Index
-    @NotNull
-    @Column(name = "instance_sequence")
-    private Integer instanceSequence;
-
-    @ForeignKey(deleteAction= ForeignKeyAction.CASCADE)
-    @ManyToOne(cascade= CascadeType.REMOVE)
-    private EntityBean entityBean;
-
-
-    @Column(name = "awaited_predicates")
-    @Lob
-    private byte[] awaitedPredicates;
-
-    @Column(name = "properties")
-    @Lob
-    private byte[] properties;
-
-
-    public String getId() {
-        return id;
-    }
-
-    public void setId(String id) {
-        this.id = id;
-    }
-
-    public String getCluster() {
-        return cluster;
-    }
-
-    public void setCluster(String cluster) {
-        this.cluster = cluster;
-    }
-
-    public String getExternalID() {
-        return externalID;
-    }
-
-    public void setExternalID(String externalID) {
-        this.externalID = externalID;
-    }
-
-    public Timestamp getInstanceTime() {
-        return instanceTime;
-    }
-
-    public void setInstanceTime(Timestamp instanceTime) {
-        this.instanceTime = instanceTime;
-    }
-
-    public Timestamp getCreationTime() {
-        return creationTime;
-    }
-
-    public void setCreationTime(Timestamp creationTime) {
-        this.creationTime = creationTime;
-    }
-
-    public Timestamp getActualStartTime() {
-        return actualStartTime;
-    }
-
-    public void setActualStartTime(Timestamp actualStartTime) {
-        this.actualStartTime = actualStartTime;
-    }
-
-    public Timestamp getActualEndTime() {
-        return actualEndTime;
-    }
-
-    public void setActualEndTime(Timestamp actualEndTime) {
-        this.actualEndTime = actualEndTime;
-    }
-
-    public String getCurrentState() {
-        return currentState;
-    }
-
-    public void setCurrentState(String currentState) {
-        this.currentState = currentState;
-    }
-
-    public byte[] getAwaitedPredicates() {
-        return awaitedPredicates;
-    }
-
-    public void setAwaitedPredicates(byte[] awaitedPredicates) {
-        this.awaitedPredicates = awaitedPredicates;
-    }
-
-    public Integer getInstanceSequence() {
-        return instanceSequence;
-    }
-
-    public void setInstanceSequence(Integer instanceSequence) {
-        this.instanceSequence = instanceSequence;
-    }
-
-    public String getEntityId() {
-        return entityId;
-    }
-
-    public void setEntityId(String entityId) {
-        this.entityId = entityId;
-    }
-
-    public byte[] getProperties() {
-        return properties;
-    }
-
-    public void setProperties(byte[] properties) {
-        this.properties = properties;
-    }
-
-    public EntityBean getEntityBean() {
-        return entityBean;
-    }
-
-    public void setEntityBean(EntityBean entityBean) {
-        this.entityBean = entityBean;
-    }
-}

http://git-wip-us.apache.org/repos/asf/falcon/blob/de2f5c0a/scheduler/src/main/java/org/apache/falcon/state/store/jdbc/JDBCStateStore.java
----------------------------------------------------------------------
diff --git a/scheduler/src/main/java/org/apache/falcon/state/store/jdbc/JDBCStateStore.java b/scheduler/src/main/java/org/apache/falcon/state/store/jdbc/JDBCStateStore.java
index 1c07286..d2bb8c8 100644
--- a/scheduler/src/main/java/org/apache/falcon/state/store/jdbc/JDBCStateStore.java
+++ b/scheduler/src/main/java/org/apache/falcon/state/store/jdbc/JDBCStateStore.java
@@ -22,6 +22,8 @@ import org.apache.commons.lang.StringUtils;
 import org.apache.falcon.entity.v0.Entity;
 import org.apache.falcon.exception.StateStoreException;
 import org.apache.falcon.execution.ExecutionInstance;
+import org.apache.falcon.persistence.EntityBean;
+import org.apache.falcon.persistence.InstanceBean;
 import org.apache.falcon.state.EntityClusterID;
 import org.apache.falcon.state.EntityID;
 import org.apache.falcon.state.EntityState;
@@ -30,7 +32,7 @@ import org.apache.falcon.state.InstanceID;
 import org.apache.falcon.state.InstanceState;
 import org.apache.falcon.state.store.AbstractStateStore;
 import org.apache.falcon.state.store.StateStore;
-import org.apache.falcon.state.store.service.FalconJPAService;
+import org.apache.falcon.service.FalconJPAService;
 import org.apache.falcon.util.StateStoreProperties;
 import org.joda.time.DateTime;
 

http://git-wip-us.apache.org/repos/asf/falcon/blob/de2f5c0a/scheduler/src/main/java/org/apache/falcon/state/store/service/FalconJPAService.java
----------------------------------------------------------------------
diff --git a/scheduler/src/main/java/org/apache/falcon/state/store/service/FalconJPAService.java b/scheduler/src/main/java/org/apache/falcon/state/store/service/FalconJPAService.java
deleted file mode 100644
index 027a8ef..0000000
--- a/scheduler/src/main/java/org/apache/falcon/state/store/service/FalconJPAService.java
+++ /dev/null
@@ -1,171 +0,0 @@
-/**
- * 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.falcon.state.store.service;
-
-import org.apache.commons.lang.StringUtils;
-import org.apache.falcon.FalconException;
-import org.apache.falcon.service.FalconService;
-import org.apache.falcon.state.store.jdbc.EntityBean;
-import org.apache.falcon.state.store.jdbc.InstanceBean;
-import org.apache.falcon.util.StateStoreProperties;
-import org.apache.openjpa.persistence.OpenJPAEntityManagerFactorySPI;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.persistence.EntityManager;
-import javax.persistence.EntityManagerFactory;
-import javax.persistence.Persistence;
-import java.text.MessageFormat;
-import java.util.Properties;
-
-/**
- * Service that manages JPA.
- */
-public final class FalconJPAService implements FalconService {
-
-    private static final Logger LOG = LoggerFactory.getLogger(FalconJPAService.class);
-    public static final String PREFIX = "falcon.statestore.";
-
-    public static final String DB_SCHEMA = PREFIX + "schema.name";
-    public static final String URL = PREFIX + "jdbc.url";
-    public static final String DRIVER = PREFIX + "jdbc.driver";
-    public static final String USERNAME = PREFIX + "jdbc.username";
-    public static final String PASSWORD = PREFIX + "jdbc.password";
-    public static final String CONN_DATA_SOURCE = PREFIX + "connection.data.source";
-    public static final String CONN_PROPERTIES = PREFIX + "connection.properties";
-    public static final String MAX_ACTIVE_CONN = PREFIX + "pool.max.active.conn";
-    public static final String CREATE_DB_SCHEMA = PREFIX + "create.db.schema";
-    public static final String VALIDATE_DB_CONN = PREFIX + "validate.db.connection";
-    public static final String VALIDATE_DB_CONN_EVICTION_INTERVAL = PREFIX + "validate.db.connection.eviction.interval";
-    public static final String VALIDATE_DB_CONN_EVICTION_NUM = PREFIX + "validate.db.connection.eviction.num";
-
-    private EntityManagerFactory entityManagerFactory;
-    // Persistent Unit which is defined in persistence.xml
-    private String persistenceUnit;
-    private static final FalconJPAService FALCON_JPA_SERVICE = new FalconJPAService();
-
-    private FalconJPAService() {
-    }
-
-    public static FalconJPAService get() {
-        return FALCON_JPA_SERVICE;
-    }
-
-    public EntityManagerFactory getEntityManagerFactory() {
-        return entityManagerFactory;
-    }
-
-    public void setPersistenceUnit(String dbType) {
-        if (StringUtils.isEmpty(dbType)) {
-            throw new IllegalArgumentException(" DB type cannot be null or empty");
-        }
-        dbType = dbType.split(":")[0];
-        this.persistenceUnit = "falcon-" + dbType;
-    }
-
-    @Override
-    public String getName() {
-        return this.getClass().getSimpleName();
-    }
-
-    @Override
-    public void init() throws FalconException {
-        Properties props = getPropsforStore();
-        entityManagerFactory = Persistence.
-                createEntityManagerFactory(persistenceUnit, props);
-        EntityManager entityManager = getEntityManager();
-        entityManager.find(EntityBean.class, 1);
-        entityManager.find(InstanceBean.class, 1);
-        LOG.info("All entities initialized");
-
-        // need to use a pseudo no-op transaction so all entities, datasource
-        // and connection pool are initialized one time only
-        entityManager.getTransaction().begin();
-        OpenJPAEntityManagerFactorySPI spi = (OpenJPAEntityManagerFactorySPI) entityManagerFactory;
-        // Mask the password with '***'
-        String logMsg = spi.getConfiguration().getConnectionProperties().replaceAll("Password=.*?,", "Password=***,");
-        LOG.info("JPA configuration: {0}", logMsg);
-        entityManager.getTransaction().commit();
-        entityManager.close();
-    }
-
-    private Properties getPropsforStore() throws FalconException {
-        String dbSchema = StateStoreProperties.get().getProperty(DB_SCHEMA);
-        String url = StateStoreProperties.get().getProperty(URL);
-        String driver = StateStoreProperties.get().getProperty(DRIVER);
-        String user = StateStoreProperties.get().getProperty(USERNAME);
-        String password = StateStoreProperties.get().getProperty(PASSWORD).trim();
-        String maxConn = StateStoreProperties.get().getProperty(MAX_ACTIVE_CONN).trim();
-        String dataSource = StateStoreProperties.get().getProperty(CONN_DATA_SOURCE);
-        String connPropsConfig = StateStoreProperties.get().getProperty(CONN_PROPERTIES);
-        boolean autoSchemaCreation = Boolean.parseBoolean(StateStoreProperties.get().getProperty(CREATE_DB_SCHEMA,
-                "false"));
-        boolean validateDbConn = Boolean.parseBoolean(StateStoreProperties.get().getProperty(VALIDATE_DB_CONN, "true"));
-        String evictionInterval = StateStoreProperties.get().getProperty(VALIDATE_DB_CONN_EVICTION_INTERVAL).trim();
-        String evictionNum = StateStoreProperties.get().getProperty(VALIDATE_DB_CONN_EVICTION_NUM).trim();
-
-        if (!url.startsWith("jdbc:")) {
-            throw new FalconException("invalid JDBC URL, must start with 'jdbc:'" + url);
-        }
-        String dbType = url.substring("jdbc:".length());
-        if (dbType.indexOf(":") <= 0) {
-            throw new FalconException("invalid JDBC URL, missing vendor 'jdbc:[VENDOR]:...'" + url);
-        }
-        setPersistenceUnit(dbType);
-        String connProps = "DriverClassName={0},Url={1},Username={2},Password={3},MaxActive={4}";
-        connProps = MessageFormat.format(connProps, driver, url, user, password, maxConn);
-        Properties props = new Properties();
-        if (autoSchemaCreation) {
-            connProps += ",TestOnBorrow=false,TestOnReturn=false,TestWhileIdle=false";
-            props.setProperty("openjpa.jdbc.SynchronizeMappings", "buildSchema(ForeignKeys=true)");
-        } else if (validateDbConn) {
-            // validation can be done only if the schema already exist, else a
-            // connection cannot be obtained to create the schema.
-            String interval = "timeBetweenEvictionRunsMillis=" + evictionInterval;
-            String num = "numTestsPerEvictionRun=" + evictionNum;
-            connProps += ",TestOnBorrow=true,TestOnReturn=true,TestWhileIdle=true," + interval + "," + num;
-            connProps += ",ValidationQuery=select 1";
-            connProps = MessageFormat.format(connProps, dbSchema);
-        } else {
-            connProps += ",TestOnBorrow=false,TestOnReturn=false,TestWhileIdle=false";
-        }
-        if (connPropsConfig != null) {
-            connProps += "," + connPropsConfig;
-        }
-        props.setProperty("openjpa.ConnectionProperties", connProps);
-        props.setProperty("openjpa.ConnectionDriverName", dataSource);
-        return props;
-    }
-
-    @Override
-    public void destroy() throws FalconException {
-        if (entityManagerFactory.isOpen()) {
-            entityManagerFactory.close();
-        }
-    }
-
-
-    /**
-     * Return an EntityManager. Used by the StoreService.
-     *
-     * @return an entity manager
-     */
-    public EntityManager getEntityManager() {
-        return getEntityManagerFactory().createEntityManager();
-    }
-}

http://git-wip-us.apache.org/repos/asf/falcon/blob/de2f5c0a/scheduler/src/main/java/org/apache/falcon/tools/FalconStateStoreDBCLI.java
----------------------------------------------------------------------
diff --git a/scheduler/src/main/java/org/apache/falcon/tools/FalconStateStoreDBCLI.java b/scheduler/src/main/java/org/apache/falcon/tools/FalconStateStoreDBCLI.java
deleted file mode 100644
index 6de9f7d..0000000
--- a/scheduler/src/main/java/org/apache/falcon/tools/FalconStateStoreDBCLI.java
+++ /dev/null
@@ -1,436 +0,0 @@
-/**
- * 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.falcon.tools;
-
-import org.apache.commons.cli.CommandLine;
-import org.apache.commons.cli.Option;
-import org.apache.commons.cli.Options;
-import org.apache.commons.cli.ParseException;
-import org.apache.falcon.cli.CLIParser;
-import org.apache.falcon.state.store.service.FalconJPAService;
-import org.apache.falcon.util.BuildProperties;
-import org.apache.falcon.util.StateStoreProperties;
-
-import java.io.File;
-import java.io.FileWriter;
-import java.io.PrintWriter;
-import java.sql.Connection;
-import java.sql.DriverManager;
-import java.sql.ResultSet;
-import java.sql.Statement;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-/**
- * Command Line utility for Table Creation, Update.
- */
-public class FalconStateStoreDBCLI {
-    public static final String HELP_CMD = "help";
-    public static final String VERSION_CMD = "version";
-    public static final String CREATE_CMD = "create";
-    public static final String SQL_FILE_OPT = "sqlfile";
-    public static final String RUN_OPT = "run";
-    public static final String UPGRADE_CMD = "upgrade";
-
-    // Represents whether DB instance exists or not.
-    private boolean instanceExists;
-    private static final String[] FALCON_HELP =
-    {"Falcon DB initialization tool currently supports Derby DB/ Mysql/ PostgreSQL"};
-
-    public static void main(String[] args) {
-        new FalconStateStoreDBCLI().run(args);
-    }
-
-    public FalconStateStoreDBCLI() {
-        instanceExists = false;
-    }
-
-    protected Options getOptions() {
-        Option sqlfile = new Option(SQL_FILE_OPT, true,
-                "Generate SQL script instead of creating/upgrading the DB schema");
-        Option run = new Option(RUN_OPT, false, "Confirmation option regarding DB schema creation/upgrade");
-        Options options = new Options();
-        options.addOption(sqlfile);
-        options.addOption(run);
-        return options;
-    }
-
-    public synchronized int run(String[] args) {
-        if (instanceExists) {
-            throw new IllegalStateException("CLI instance already used");
-        }
-        instanceExists = true;
-
-        CLIParser parser = new CLIParser("falcondb", FALCON_HELP);
-        parser.addCommand(HELP_CMD, "", "Display usage for all commands or specified command", new Options(), false);
-        parser.addCommand(VERSION_CMD, "", "Show Falcon DB version information", new Options(), false);
-        parser.addCommand(CREATE_CMD, "", "Create Falcon DB schema", getOptions(), false);
-        parser.addCommand(UPGRADE_CMD, "", "Upgrade Falcon DB schema", getOptions(), false);
-
-        try {
-            CLIParser.Command command = parser.parse(args);
-            if (command.getName().equals(HELP_CMD)) {
-                parser.showHelp();
-            } else if (command.getName().equals(VERSION_CMD)) {
-                showVersion();
-            } else {
-                if (!command.getCommandLine().hasOption(SQL_FILE_OPT)
-                        && !command.getCommandLine().hasOption(RUN_OPT)) {
-                    throw new Exception("'-sqlfile <FILE>' or '-run' options must be specified");
-                }
-                CommandLine commandLine = command.getCommandLine();
-                String sqlFile = (commandLine.hasOption(SQL_FILE_OPT))
-                        ? commandLine.getOptionValue(SQL_FILE_OPT)
-                        : File.createTempFile("falcondb-", ".sql").getAbsolutePath();
-                boolean run = commandLine.hasOption(RUN_OPT);
-                if (command.getName().equals(CREATE_CMD)) {
-                    createDB(sqlFile, run);
-                } else if (command.getName().equals(UPGRADE_CMD)) {
-                    upgradeDB(sqlFile, run);
-                }
-                System.out.println("The SQL commands have been written to: " + sqlFile);
-                if (!run) {
-                    System.out.println("WARN: The SQL commands have NOT been executed, you must use the '-run' option");
-                }
-            }
-            return 0;
-        } catch (ParseException ex) {
-            System.err.println("Invalid sub-command: " + ex.getMessage());
-            System.err.println();
-            System.err.println(parser.shortHelp());
-            return 1;
-        } catch (Exception ex) {
-            System.err.println();
-            System.err.println("Error: " + ex.getMessage());
-            System.err.println();
-            System.err.println("Stack trace for the error was (for debug purposes):");
-            System.err.println("--------------------------------------");
-            ex.printStackTrace(System.err);
-            System.err.println("--------------------------------------");
-            System.err.println();
-            return 1;
-        }
-    }
-
-    private void upgradeDB(String sqlFile, boolean run) throws Exception {
-        validateConnection();
-        if (!checkDBExists()) {
-            throw new Exception("Falcon DB doesn't exist");
-        }
-        String falconVersion = BuildProperties.get().getProperty("project.version");
-        String dbVersion = getFalconDBVersion();
-        if (dbVersion.compareTo(falconVersion) >= 0) {
-            System.out.println("Falcon DB already upgraded to Falcon version '" + falconVersion + "'");
-            return;
-        }
-
-        createUpgradeDB(sqlFile, run, false);
-        upgradeFalconDBVersion(sqlFile, run, falconVersion);
-
-        // any post upgrade tasks
-        if (run) {
-            System.out.println("Falcon DB has been upgraded to Falcon version '" + falconVersion + "'");
-        }
-    }
-
-
-    private void upgradeFalconDBVersion(String sqlFile, boolean run, String version) throws Exception {
-        String updateDBVersion = "update FALCON_DB_PROPS set data='" + version + "' where name='db.version'";
-        PrintWriter writer = new PrintWriter(new FileWriter(sqlFile, true));
-        writer.println();
-        writer.println(updateDBVersion);
-        writer.close();
-        System.out.println("Upgrade db.version in FALCON_DB_PROPS table to " + version);
-        if (run) {
-            Connection conn = createConnection();
-            Statement st = null;
-            try {
-                conn.setAutoCommit(true);
-                st = conn.createStatement();
-                st.executeUpdate(updateDBVersion);
-                st.close();
-            } catch (Exception ex) {
-                throw new Exception("Could not upgrade db.version in FALCON_DB_PROPS table: " + ex.toString(), ex);
-            } finally {
-                closeStatement(st);
-                conn.close();
-            }
-        }
-        System.out.println("DONE");
-    }
-
-    private static final String GET_FALCON_DB_VERSION = "select data from FALCON_DB_PROPS where name = 'db.version'";
-
-    private String getFalconDBVersion() throws Exception {
-        String version;
-        System.out.println("Get Falcon DB version");
-        Connection conn = createConnection();
-        Statement st = null;
-        ResultSet rs = null;
-        try {
-            st = conn.createStatement();
-            rs = st.executeQuery(GET_FALCON_DB_VERSION);
-            if (rs.next()) {
-                version = rs.getString(1);
-            } else {
-                throw new Exception("ERROR: Could not find Falcon DB 'db.version' in FALCON_DB_PROPS table");
-            }
-        } catch (Exception ex) {
-            throw new Exception("ERROR: Could not query FALCON_DB_PROPS table: " + ex.toString(), ex);
-        } finally {
-            closeResultSet(rs);
-            closeStatement(st);
-            conn.close();
-        }
-        System.out.println("DONE");
-        return version;
-    }
-
-
-    private Map<String, String> getJdbcConf() throws Exception {
-        Map<String, String> jdbcConf = new HashMap<String, String>();
-        jdbcConf.put("driver", StateStoreProperties.get().getProperty(FalconJPAService.DRIVER));
-        String url = StateStoreProperties.get().getProperty(FalconJPAService.URL);
-        jdbcConf.put("url", url);
-        jdbcConf.put("user", StateStoreProperties.get().getProperty(FalconJPAService.USERNAME));
-        jdbcConf.put("password", StateStoreProperties.get().getProperty(FalconJPAService.PASSWORD));
-        String dbType = url.substring("jdbc:".length());
-        if (dbType.indexOf(":") <= 0) {
-            throw new RuntimeException("Invalid JDBC URL, missing vendor 'jdbc:[VENDOR]:...'");
-        }
-        dbType = dbType.substring(0, dbType.indexOf(":"));
-        jdbcConf.put("dbtype", dbType);
-        return jdbcConf;
-    }
-
-    private String[] createMappingToolArguments(String sqlFile) throws Exception {
-        Map<String, String> conf = getJdbcConf();
-        List<String> args = new ArrayList<String>();
-        args.add("-schemaAction");
-        args.add("add");
-        args.add("-p");
-        args.add("persistence.xml#falcon-" + conf.get("dbtype"));
-        args.add("-connectionDriverName");
-        args.add(conf.get("driver"));
-        args.add("-connectionURL");
-        args.add(conf.get("url"));
-        args.add("-connectionUserName");
-        args.add(conf.get("user"));
-        args.add("-connectionPassword");
-        args.add(conf.get("password"));
-        if (sqlFile != null) {
-            args.add("-sqlFile");
-            args.add(sqlFile);
-        }
-        args.add("-indexes");
-        args.add("true");
-        args.add("org.apache.falcon.state.store.jdbc.EntityBean");
-        args.add("org.apache.falcon.state.store.jdbc.InstanceBean");
-        return args.toArray(new String[args.size()]);
-    }
-
-    private void createDB(String sqlFile, boolean run) throws Exception {
-        validateConnection();
-        if (checkDBExists()) {
-            return;
-        }
-
-        verifyFalconPropsTable(false);
-        createUpgradeDB(sqlFile, run, true);
-        createFalconPropsTable(sqlFile, run, BuildProperties.get().getProperty("project.version"));
-        if (run) {
-            System.out.println("Falcon DB has been created for Falcon version '"
-                    + BuildProperties.get().getProperty("project.version") + "'");
-        }
-    }
-
-    private static final String CREATE_FALCON_DB_PROPS =
-            "create table FALCON_DB_PROPS (name varchar(100), data varchar(100))";
-
-    private void createFalconPropsTable(String sqlFile, boolean run, String version) throws Exception {
-        String insertDbVerion = "insert into FALCON_DB_PROPS (name, data) values ('db.version', '" + version + "')";
-
-        PrintWriter writer = new PrintWriter(new FileWriter(sqlFile, true));
-        writer.println();
-        writer.println(CREATE_FALCON_DB_PROPS);
-        writer.println(insertDbVerion);
-        writer.close();
-        System.out.println("Create FALCON_DB_PROPS table");
-        if (run) {
-            Connection conn = createConnection();
-            Statement st = null;
-            try {
-                conn.setAutoCommit(true);
-                st = conn.createStatement();
-                st.executeUpdate(CREATE_FALCON_DB_PROPS);
-                st.executeUpdate(insertDbVerion);
-                st.close();
-            } catch (Exception ex) {
-                closeStatement(st);
-                throw new Exception("Could not create FALCON_DB_PROPS table: " + ex.toString(), ex);
-            } finally {
-                conn.close();
-            }
-        }
-        System.out.println("DONE");
-    }
-
-    private static final String FALCON_DB_PROPS_EXISTS = "select count(*) from FALCON_DB_PROPS";
-
-    private boolean verifyFalconPropsTable(boolean exists) throws Exception {
-        System.out.println((exists) ? "Check FALCON_DB_PROPS table exists"
-                : "Checking FALCON_DB_PROPS table does not exist");
-        boolean tableExists;
-        Connection conn = createConnection();
-        Statement st = null;
-        ResultSet rs = null;
-        try {
-            st = conn.createStatement();
-            rs = st.executeQuery(FALCON_DB_PROPS_EXISTS);
-            rs.next();
-            tableExists = true;
-        } catch (Exception ex) {
-            tableExists = false;
-        } finally {
-            closeResultSet(rs);
-            closeStatement(st);
-            conn.close();
-        }
-        if (tableExists != exists) {
-            throw new Exception("FALCON_DB_PROPS_TABLE table " + ((exists) ? "does not exist" : "exists"));
-        }
-        System.out.println("DONE");
-        return tableExists;
-    }
-
-    private void closeResultSet(ResultSet rs) {
-        try {
-            if (rs != null) {
-                rs.close();
-            }
-        } catch (Exception e) {
-            System.out.println("Unable to close ResultSet " + rs);
-        }
-    }
-
-    private void closeStatement(Statement st) throws Exception {
-        try {
-            if (st != null) {
-                st.close();
-            }
-        } catch (Exception e) {
-            System.out.println("Unable to close SQL Statement " + st);
-            throw new Exception(e);
-        }
-    }
-
-    private Connection createConnection() throws Exception {
-        Map<String, String> conf = getJdbcConf();
-        Class.forName(conf.get("driver")).newInstance();
-        return DriverManager.getConnection(conf.get("url"), conf.get("user"), conf.get("password"));
-    }
-
-    private void validateConnection() throws Exception {
-        System.out.println("Validating DB Connection");
-        try {
-            createConnection().close();
-            System.out.println("DONE");
-        } catch (Exception ex) {
-            throw new Exception("Could not connect to the database: " + ex.toString(), ex);
-        }
-    }
-
-    private static final String ENTITY_STATUS_QUERY =
-            "select count(*) from ENTITIES where current_state IN ('RUNNING', 'SUSPENDED')";
-    private static final String INSTANCE_STATUS_QUERY =
-            "select count(*) from INSTANCES where current_state IN ('RUNNING', 'SUSPENDED')";
-
-    private boolean checkDBExists() throws Exception {
-        boolean schemaExists;
-        Connection conn = createConnection();
-        ResultSet rs =  null;
-        Statement st = null;
-        try {
-            st = conn.createStatement();
-            rs = st.executeQuery(ENTITY_STATUS_QUERY);
-            rs.next();
-            schemaExists = true;
-        } catch (Exception ex) {
-            schemaExists = false;
-        } finally {
-            closeResultSet(rs);
-            closeStatement(st);
-            conn.close();
-        }
-        System.out.println("DB schema " + ((schemaExists) ? "exists" : "does not exist"));
-        return schemaExists;
-    }
-
-    private void createUpgradeDB(String sqlFile, boolean run, boolean create) throws Exception {
-        System.out.println((create) ? "Create SQL schema" : "Upgrade SQL schema");
-        String[] args = createMappingToolArguments(sqlFile);
-        org.apache.openjpa.jdbc.meta.MappingTool.main(args);
-        if (run) {
-            args = createMappingToolArguments(null);
-            org.apache.openjpa.jdbc.meta.MappingTool.main(args);
-        }
-        System.out.println("DONE");
-    }
-
-    private void showVersion() throws Exception {
-        System.out.println("Falcon Server version: "
-                + BuildProperties.get().getProperty("project.version"));
-        validateConnection();
-        if (!checkDBExists()) {
-            throw new Exception("Falcon DB doesn't exist");
-        }
-        try {
-            verifyFalconPropsTable(true);
-        } catch (Exception ex) {
-            throw new Exception("ERROR: It seems this Falcon DB was never upgraded with the 'falcondb' tool");
-        }
-        showFalconPropsInfo();
-    }
-
-    private static final String GET_FALCON_PROPS_INFO = "select name, data from FALCON_DB_PROPS order by name";
-
-    private void showFalconPropsInfo() throws Exception {
-        Connection conn = createConnection();
-        Statement st = null;
-        ResultSet rs = null;
-        try {
-            System.out.println("Falcon DB Version Information");
-            System.out.println("--------------------------------------");
-            st = conn.createStatement();
-            rs = st.executeQuery(GET_FALCON_PROPS_INFO);
-            while (rs.next()) {
-                System.out.println(rs.getString(1) + ": " + rs.getString(2));
-            }
-            System.out.println("--------------------------------------");
-        } catch (Exception ex) {
-            throw new Exception("ERROR querying FALCON_DB_PROPS table: " + ex.toString(), ex);
-        } finally {
-            closeResultSet(rs);
-            closeStatement(st);
-            conn.close();
-        }
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/falcon/blob/de2f5c0a/scheduler/src/main/resources/META-INF/persistence.xml
----------------------------------------------------------------------
diff --git a/scheduler/src/main/resources/META-INF/persistence.xml b/scheduler/src/main/resources/META-INF/persistence.xml
deleted file mode 100644
index c2ef794..0000000
--- a/scheduler/src/main/resources/META-INF/persistence.xml
+++ /dev/null
@@ -1,104 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
-  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.
--->
-<persistence xmlns="http://java.sun.com/xml/ns/persistence"
-             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-             version="1.0">
-
-    <persistence-unit name="falcon-derby" transaction-type="RESOURCE_LOCAL">
-        <provider>org.apache.openjpa.persistence.PersistenceProviderImpl</provider>
-
-        <class>org.apache.falcon.state.store.jdbc.EntityBean</class>
-        <class>org.apache.falcon.state.store.jdbc.InstanceBean</class>
-
-        <properties>
-            <property name="openjpa.ConnectionDriverName" value="org.apache.commons.dbcp.BasicDataSource"/>
-
-            <property name="openjpa.ConnectionProperties" value="**INVALID**"/> <!--Set by StoreService at init time -->
-
-            <property name="openjpa.MetaDataFactory"
-                      value="jpa(Types=org.apache.falcon.state.store.EntityBean;
-                org.apache.falcon.state.store.InstanceBean)"></property>
-
-            <property name="openjpa.DetachState" value="fetch-groups(DetachedStateField=true)"/>
-            <property name="openjpa.LockManager" value="pessimistic"/>
-            <property name="openjpa.ReadLockLevel" value="read"/>
-            <property name="openjpa.WriteLockLevel" value="write"/>
-            <property name="openjpa.jdbc.TransactionIsolation" value="read-committed"/> <!--CUSTOM-->
-            <property name="openjpa.jdbc.DBDictionary" value="batchLimit=50"/>
-            <property name="openjpa.jdbc.DBDictionary" value="TimestampTypeName=TIMESTAMP"/>
-            <property name="openjpa.RuntimeUnenhancedClasses" value="unsupported"/>
-            <property name="openjpa.Log" value="log4j"/>
-        </properties>
-    </persistence-unit>
-
-    <persistence-unit name="falcon-mysql" transaction-type="RESOURCE_LOCAL">
-        <provider>org.apache.openjpa.persistence.PersistenceProviderImpl</provider>
-
-        <class>org.apache.falcon.state.store.jdbc.EntityBean</class>
-        <class>org.apache.falcon.state.store.jdbc.InstanceBean</class>
-
-        <properties>
-            <property name="openjpa.ConnectionDriverName" value="org.apache.commons.dbcp.BasicDataSource"/>
-
-            <property name="openjpa.ConnectionProperties" value="**INVALID**"/> <!--Set by StoreService at init time -->
-
-            <property name="openjpa.MetaDataFactory"
-                      value="jpa(Types=org.apache.falcon.state.store.EntityBean;
-                org.apache.falcon.state.store.InstanceBean)"></property>
-
-            <property name="openjpa.DetachState" value="fetch-groups(DetachedStateField=true)"/>
-            <property name="openjpa.LockManager" value="pessimistic"/>
-            <property name="openjpa.ReadLockLevel" value="read"/>
-            <property name="openjpa.WriteLockLevel" value="write"/>
-            <property name="openjpa.jdbc.TransactionIsolation" value="repeatable-read"/> <!--CUSTOM-->
-            <property name="openjpa.jdbc.DBDictionary" value="batchLimit=50"/>
-            <property name="openjpa.jdbc.DBDictionary" value="TimestampTypeName=TIMESTAMP"/>
-            <property name="openjpa.RuntimeUnenhancedClasses" value="unsupported"/>
-            <property name="openjpa.Log" value="log4j"/>
-        </properties>
-    </persistence-unit>
-
-    <persistence-unit name="falcon-postgresql" transaction-type="RESOURCE_LOCAL">
-        <provider>org.apache.openjpa.persistence.PersistenceProviderImpl</provider>
-
-        <class>org.apache.falcon.state.store.jdbc.EntityBean</class>
-        <class>org.apache.falcon.state.store.jdbc.InstanceBean</class>
-
-        <properties>
-            <property name="openjpa.ConnectionDriverName" value="org.apache.commons.dbcp.BasicDataSource"/>
-
-            <property name="openjpa.ConnectionProperties" value="**INVALID**"/> <!--Set by StoreService at init time -->
-
-            <property name="openjpa.MetaDataFactory"
-                      value="jpa(Types=org.apache.falcon.state.store.EntityBean;
-                org.apache.falcon.state.store.InstanceBean)"></property>
-
-            <property name="openjpa.DetachState" value="fetch-groups(DetachedStateField=true)"/>
-            <property name="openjpa.LockManager" value="pessimistic"/>
-            <property name="openjpa.ReadLockLevel" value="read"/>
-            <property name="openjpa.WriteLockLevel" value="write"/>
-            <property name="openjpa.jdbc.TransactionIsolation" value="repeatable-read"/> <!--CUSTOM-->
-            <property name="openjpa.jdbc.DBDictionary" value="batchLimit=50"/>
-            <property name="openjpa.jdbc.DBDictionary" value="TimestampTypeName=TIMESTAMP"/>
-            <property name="openjpa.RuntimeUnenhancedClasses" value="unsupported"/>
-            <property name="openjpa.Log" value="log4j"/>
-        </properties>
-    </persistence-unit>
-
-</persistence>

http://git-wip-us.apache.org/repos/asf/falcon/blob/de2f5c0a/scheduler/src/test/java/org/apache/falcon/execution/FalconExecutionServiceTest.java
----------------------------------------------------------------------
diff --git a/scheduler/src/test/java/org/apache/falcon/execution/FalconExecutionServiceTest.java b/scheduler/src/test/java/org/apache/falcon/execution/FalconExecutionServiceTest.java
index 417ec3e..437c5f5 100644
--- a/scheduler/src/test/java/org/apache/falcon/execution/FalconExecutionServiceTest.java
+++ b/scheduler/src/test/java/org/apache/falcon/execution/FalconExecutionServiceTest.java
@@ -45,7 +45,7 @@ import org.apache.falcon.state.InstanceID;
 import org.apache.falcon.state.InstanceState;
 import org.apache.falcon.state.store.AbstractStateStore;
 import org.apache.falcon.state.store.StateStore;
-import org.apache.falcon.state.store.service.FalconJPAService;
+import org.apache.falcon.service.FalconJPAService;
 import org.apache.falcon.util.StartupProperties;
 import org.apache.falcon.workflow.engine.DAGEngine;
 import org.apache.falcon.workflow.engine.DAGEngineFactory;

http://git-wip-us.apache.org/repos/asf/falcon/blob/de2f5c0a/scheduler/src/test/java/org/apache/falcon/state/AbstractSchedulerTestBase.java
----------------------------------------------------------------------
diff --git a/scheduler/src/test/java/org/apache/falcon/state/AbstractSchedulerTestBase.java b/scheduler/src/test/java/org/apache/falcon/state/AbstractSchedulerTestBase.java
index 155be69..cd99049 100644
--- a/scheduler/src/test/java/org/apache/falcon/state/AbstractSchedulerTestBase.java
+++ b/scheduler/src/test/java/org/apache/falcon/state/AbstractSchedulerTestBase.java
@@ -18,7 +18,7 @@
 package org.apache.falcon.state;
 
 import org.apache.falcon.entity.AbstractTestBase;
-import org.apache.falcon.state.store.service.FalconJPAService;
+import org.apache.falcon.service.FalconJPAService;
 import org.apache.falcon.tools.FalconStateStoreDBCLI;
 import org.apache.falcon.util.StateStoreProperties;
 import org.apache.hadoop.conf.Configuration;

http://git-wip-us.apache.org/repos/asf/falcon/blob/de2f5c0a/scheduler/src/test/java/org/apache/falcon/state/service/TestFalconJPAService.java
----------------------------------------------------------------------
diff --git a/scheduler/src/test/java/org/apache/falcon/state/service/TestFalconJPAService.java b/scheduler/src/test/java/org/apache/falcon/state/service/TestFalconJPAService.java
index ecd5293..3e186dd 100644
--- a/scheduler/src/test/java/org/apache/falcon/state/service/TestFalconJPAService.java
+++ b/scheduler/src/test/java/org/apache/falcon/state/service/TestFalconJPAService.java
@@ -19,7 +19,7 @@ package org.apache.falcon.state.service;
 
 import org.apache.falcon.FalconException;
 import org.apache.falcon.state.AbstractSchedulerTestBase;
-import org.apache.falcon.state.store.service.FalconJPAService;
+import org.apache.falcon.service.FalconJPAService;
 import org.testng.Assert;
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;

http://git-wip-us.apache.org/repos/asf/falcon/blob/de2f5c0a/scheduler/src/test/java/org/apache/falcon/state/service/store/TestJDBCStateStore.java
----------------------------------------------------------------------
diff --git a/scheduler/src/test/java/org/apache/falcon/state/service/store/TestJDBCStateStore.java b/scheduler/src/test/java/org/apache/falcon/state/service/store/TestJDBCStateStore.java
index d597e27..bf5c142 100644
--- a/scheduler/src/test/java/org/apache/falcon/state/service/store/TestJDBCStateStore.java
+++ b/scheduler/src/test/java/org/apache/falcon/state/service/store/TestJDBCStateStore.java
@@ -44,7 +44,7 @@ import org.apache.falcon.state.InstanceState;
 import org.apache.falcon.state.store.jdbc.BeanMapperUtil;
 import org.apache.falcon.state.store.jdbc.JDBCStateStore;
 import org.apache.falcon.state.store.StateStore;
-import org.apache.falcon.state.store.service.FalconJPAService;
+import org.apache.falcon.service.FalconJPAService;
 import org.apache.falcon.util.StartupProperties;
 import org.apache.falcon.workflow.engine.DAGEngine;
 import org.apache.falcon.workflow.engine.DAGEngineFactory;

http://git-wip-us.apache.org/repos/asf/falcon/blob/de2f5c0a/scheduler/src/test/resources/startup.properties
----------------------------------------------------------------------
diff --git a/scheduler/src/test/resources/startup.properties b/scheduler/src/test/resources/startup.properties
index 7160bb2..6216b70 100644
--- a/scheduler/src/test/resources/startup.properties
+++ b/scheduler/src/test/resources/startup.properties
@@ -41,7 +41,7 @@
                         org.apache.falcon.notification.service.impl.AlarmService,\
                         org.apache.falcon.notification.service.impl.DataAvailabilityService,\
                         org.apache.falcon.execution.FalconExecutionService,\
-                        org.apache.falcon.state.store.service.FalconJPAService
+                        org.apache.falcon.service.FalconJPAService
 
 ##### Falcon Configuration Store Change listeners #####
 *.configstore.listeners=org.apache.falcon.entity.v0.EntityGraph,\
@@ -121,7 +121,8 @@ debug.libext.process.paths=${falcon.libext}
 *.falcon.http.authentication.simple.anonymous.allowed=false
 
 # Indicates the Kerberos principal to be used for HTTP endpoint.
-# The principal MUST start with 'HTTP/' as per Kerberos HTTP SPNEGO specification.
+# The principal MUST
+#rt with 'HTTP/' as per Kerberos HTTP SPNEGO specification.
 *.falcon.http.authentication.kerberos.principal=
 
 # Location of the keytab file with the credentials for the HTTP principal.

http://git-wip-us.apache.org/repos/asf/falcon/blob/de2f5c0a/scheduler/src/test/resources/statestore.properties
----------------------------------------------------------------------
diff --git a/scheduler/src/test/resources/statestore.properties b/scheduler/src/test/resources/statestore.properties
index 2ae642f..e7a08fc 100644
--- a/scheduler/src/test/resources/statestore.properties
+++ b/scheduler/src/test/resources/statestore.properties
@@ -18,7 +18,7 @@
 
 
 *.domain=debug
-######## StateStore Properties #####
+######### StateStore Properties #####
 *.falcon.state.store.impl=org.apache.falcon.state.store.jdbc.JDBCStateStore
 *.falcon.statestore.jdbc.driver=org.apache.derby.jdbc.EmbeddedDriver
 *.falcon.statestore.jdbc.url=jdbc:derby:target/test-data/data.db;create=true

http://git-wip-us.apache.org/repos/asf/falcon/blob/de2f5c0a/src/build/findbugs-exclude.xml
----------------------------------------------------------------------
diff --git a/src/build/findbugs-exclude.xml b/src/build/findbugs-exclude.xml
index 78f2fd0..a6766df 100644
--- a/src/build/findbugs-exclude.xml
+++ b/src/build/findbugs-exclude.xml
@@ -38,16 +38,37 @@
     </Match>
 
     <Match>
-        <Class name="org.apache.falcon.state.store.jdbc.EntityBean" />
+        <Class name="org.apache.falcon.persistence.EntityBean" />
         <Bug pattern="NP_BOOLEAN_RETURN_NULL" />
     </Match>
 
     <Match>
-        <Class name="org.apache.falcon.state.store.jdbc.InstanceBean" />
+        <Class name="org.apache.falcon.persistence.InstanceBean" />
         <Bug pattern="NP_BOOLEAN_RETURN_NULL" />
     </Match>
 
     <Match>
+        <Class name="org.apache.falcon.persistence.PendingInstanceBean" />
+        <Bug pattern="NP_BOOLEAN_RETURN_NULL,UWF_UNWRITTEN_FIELD" />
+    </Match>
+
+    <!--<Match>-->
+        <!--<Class name="org.apache.falcon.persistence.PendingInstanceBean" />-->
+        <!--<Bug pattern="UWF_UNWRITTEN_FIELD" />-->
+    <!--</Match>-->
+
+
+    <Match>
+        <Class name="org.apache.falcon.persistence.MonitoredFeedsBean" />
+        <Bug pattern="NP_BOOLEAN_RETURN_NULL,UWF_UNWRITTEN_FIELD" />
+    </Match>
+
+    <!--<Match>-->
+        <!--<Class name="org.apache.falcon.persistence.MonitoredFeedsBean" />-->
+        <!--<Bug pattern="UWF_UNWRITTEN_FIELD" />-->
+    <!--</Match>-->
+
+    <Match>
         <Bug pattern="RV_RETURN_VALUE_IGNORED_NO_SIDE_EFFECT" />
     </Match>
 </FindBugsFilter>

http://git-wip-us.apache.org/repos/asf/falcon/blob/de2f5c0a/src/conf/startup.properties
----------------------------------------------------------------------
diff --git a/src/conf/startup.properties b/src/conf/startup.properties
index f23337b..3601e22 100644
--- a/src/conf/startup.properties
+++ b/src/conf/startup.properties
@@ -59,7 +59,7 @@
 #                        org.apache.falcon.service.ProcessSubscriberService,\
 #                        org.apache.falcon.service.FeedSLAMonitoringService,\
 #                        org.apache.falcon.service.LifecyclePolicyMap,\
-#                        org.apache.falcon.state.store.service.FalconJPAService,\
+#                        org.apache.falcon.service.FalconJPAService,\
 #                        org.apache.falcon.entity.store.ConfigurationStore,\
 #                        org.apache.falcon.rerun.service.RetryService,\
 #                        org.apache.falcon.rerun.service.LateRunService,\

http://git-wip-us.apache.org/repos/asf/falcon/blob/de2f5c0a/unit/pom.xml
----------------------------------------------------------------------
diff --git a/unit/pom.xml b/unit/pom.xml
index 7e5b073..f1ef463 100644
--- a/unit/pom.xml
+++ b/unit/pom.xml
@@ -44,6 +44,16 @@
         <dependency>
             <groupId>org.apache.oozie</groupId>
             <artifactId>oozie-core</artifactId>
+            <exclusions>
+                <exclusion>
+            <groupId>org.apache.openjpa</groupId>
+                    <artifactId>openjpa-jdbc</artifactId>
+                    </exclusion>
+                <exclusion>
+                    <groupId>org.apache.openjpa</groupId>
+                    <artifactId>openjpa-persistence</artifactId>
+                </exclusion>
+            </exclusions>
         </dependency>
 
         <dependency>

http://git-wip-us.apache.org/repos/asf/falcon/blob/de2f5c0a/unit/src/main/resources/startup.properties
----------------------------------------------------------------------
diff --git a/unit/src/main/resources/startup.properties b/unit/src/main/resources/startup.properties
index 4576e0b..4dfea31 100644
--- a/unit/src/main/resources/startup.properties
+++ b/unit/src/main/resources/startup.properties
@@ -33,6 +33,7 @@
 *.application.services=org.apache.falcon.security.AuthenticationInitializationService,\
                         org.apache.falcon.workflow.WorkflowJobEndNotificationService, \
                         org.apache.falcon.service.ProcessSubscriberService,\
+                        org.apache.falcon.service.FalconJPAService,\
                         org.apache.falcon.entity.store.ConfigurationStore,\
                         org.apache.falcon.rerun.service.RetryService,\
                         org.apache.falcon.rerun.service.LateRunService,\

http://git-wip-us.apache.org/repos/asf/falcon/blob/de2f5c0a/unit/src/test/java/org/apache/falcon/unit/TestFalconUnit.java
----------------------------------------------------------------------
diff --git a/unit/src/test/java/org/apache/falcon/unit/TestFalconUnit.java b/unit/src/test/java/org/apache/falcon/unit/TestFalconUnit.java
index aaf2b37..9b1ff2a 100644
--- a/unit/src/test/java/org/apache/falcon/unit/TestFalconUnit.java
+++ b/unit/src/test/java/org/apache/falcon/unit/TestFalconUnit.java
@@ -133,6 +133,7 @@ public class TestFalconUnit extends FalconUnitTestBase {
             ParseException, InterruptedException {
         // submit cluster and feeds
         submitClusterAndFeeds();
+
         APIResult result = submitProcess(getAbsolutePath(PROCESS), PROCESS_APP_PATH);
         assertStatus(result);
         createData(INPUT_FEED_NAME, CLUSTER_NAME, SCHEDULE_TIME, INPUT_FILE_NAME);

http://git-wip-us.apache.org/repos/asf/falcon/blob/de2f5c0a/webapp/src/test/java/org/apache/falcon/resource/AbstractSchedulerManagerJerseyIT.java
----------------------------------------------------------------------
diff --git a/webapp/src/test/java/org/apache/falcon/resource/AbstractSchedulerManagerJerseyIT.java b/webapp/src/test/java/org/apache/falcon/resource/AbstractSchedulerManagerJerseyIT.java
index 175833a..1bd4f45 100644
--- a/webapp/src/test/java/org/apache/falcon/resource/AbstractSchedulerManagerJerseyIT.java
+++ b/webapp/src/test/java/org/apache/falcon/resource/AbstractSchedulerManagerJerseyIT.java
@@ -24,7 +24,7 @@ import org.apache.falcon.client.FalconCLIException;
 import org.apache.falcon.entity.v0.EntityType;
 import org.apache.falcon.hadoop.HadoopClientFactory;
 import org.apache.falcon.state.AbstractSchedulerTestBase;
-import org.apache.falcon.state.store.service.FalconJPAService;
+import org.apache.falcon.service.FalconJPAService;
 import org.apache.falcon.unit.FalconUnitTestBase;
 import org.apache.falcon.util.StartupProperties;
 import org.apache.falcon.util.StateStoreProperties;


Mime
View raw message