asterixdb-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From xi...@apache.org
Subject asterixdb git commit: Feed Policy Refactoring
Date Wed, 05 Apr 2017 18:02:36 GMT
Repository: asterixdb
Updated Branches:
  refs/heads/master 05e4256a8 -> 246fbde6d


Feed Policy Refactoring

1. Fix the framesize inconsistency in feed dataflow.
2. Add a runtime test case for create feed with policy.
3. Fix the FeedRuntimeInputHandler flush() logic. Only flush after the
   writer is opened.
4. Refactor FeedPolicyAccessor and BuiltinPolices. Now we only have
   spill and discard polices.
5. Remove PolicyEnforcer. Merge the functionality into
   FeedPolicyAccessor.
6. Revise SocketServerInputStream. Make the expected exception more
   friendly.
7. Fixed one test case fail in change feed.
8. Refactor FeedRuntimeInputHandler consumer thread logic. Change the
   poison consumer thread mechanism.

Change-Id: Ibc10139925cfedee66d1263990ba80b94675f182
Reviewed-on: https://asterix-gerrit.ics.uci.edu/1591
Reviewed-by: abdullah alamoudi <bamousaa@gmail.com>
Integration-Tests: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Tested-by: Jenkins <jenkins@fulliautomatix.ics.uci.edu>


Project: http://git-wip-us.apache.org/repos/asf/asterixdb/repo
Commit: http://git-wip-us.apache.org/repos/asf/asterixdb/commit/246fbde6
Tree: http://git-wip-us.apache.org/repos/asf/asterixdb/tree/246fbde6
Diff: http://git-wip-us.apache.org/repos/asf/asterixdb/diff/246fbde6

Branch: refs/heads/master
Commit: 246fbde6d98de0929e610b64b5752aa354d98e20
Parents: 05e4256
Author: Xikui Wang <xkkwww@gmail.com>
Authored: Tue Apr 4 17:53:51 2017 -0700
Committer: Xikui Wang <xkkwww@gmail.com>
Committed: Wed Apr 5 11:02:14 2017 -0700

----------------------------------------------------------------------
 .../apache/asterix/utils/FeedOperations.java    |   5 +-
 .../create-feed-with-policy.1.ddl.aql           |  58 +++++++++
 .../create-feed-with-policy.2.update.aql        |  32 +++++
 .../create-feed-with-policy.3.server.aql        |  26 ++++
 .../create-feed-with-policy.4.sleep.aql         |  26 ++++
 .../create-feed-with-policy.5.update.aql        |  29 +++++
 .../create-feed-with-policy.6.query.aql         |  31 +++++
 .../create-feed-with-policy.7.server.aql        |  27 +++++
 .../create-feed-with-policy.8.ddl.aql           |  28 +++++
 .../feeds/feeds_08/feeds_08.2.update.sqlpp      |   2 +-
 .../feeds/feeds_09/feeds_09.2.update.sqlpp      |   2 +-
 .../feeds/feeds_10/feeds_10.2.update.sqlpp      |   2 +-
 .../feeds/feeds_11/feeds_11.2.update.sqlpp      |   2 +-
 .../feeds/feeds_12/feeds_12.2.update.sqlpp      |   2 +-
 .../issue_230_feeds.2.update.sqlpp              |   2 +-
 .../issue_251_dataset_hint_7.2.update.sqlpp     |   2 +-
 .../create-feed-with-policy.1.adm               |  10 ++
 .../results/feeds/feeds_03/feeds_03.1.adm       |   2 +-
 .../feed/dataflow/FeedRuntimeInputHandler.java  | 118 +++++++++----------
 .../feed/policy/FeedPolicyAccessor.java         |  83 +------------
 .../feed/policy/FeedPolicyEnforcer.java         |  48 --------
 .../input/stream/SocketServerInputStream.java   |   6 +-
 .../FeedCollectOperatorNodePushable.java        |   2 +-
 .../operators/FeedMetaComputeNodePushable.java  |  11 +-
 .../operators/FeedMetaStoreNodePushable.java    |  12 +-
 .../asterix/external/util/FeedConstants.java    |   4 -
 .../external/feed/test/InputHandlerTest.java    |  26 +++-
 .../metadata/feeds/BuiltinFeedPolicies.java     | 116 ++----------------
 28 files changed, 391 insertions(+), 323 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/asterixdb/blob/246fbde6/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/FeedOperations.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/FeedOperations.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/FeedOperations.java
index 874cbb1..8b3b020 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/FeedOperations.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/FeedOperations.java
@@ -32,6 +32,7 @@ import org.apache.asterix.active.ActiveRuntimeId;
 import org.apache.asterix.active.EntityId;
 import org.apache.asterix.active.message.ActiveManagerMessage;
 import org.apache.asterix.app.translator.DefaultStatementExecutorFactory;
+import org.apache.asterix.common.config.CompilerProperties;
 import org.apache.asterix.common.context.IStorageComponentProvider;
 import org.apache.asterix.common.dataflow.LSMTreeInsertDeleteOperatorDescriptor;
 import org.apache.asterix.common.exceptions.ACIDException;
@@ -110,13 +111,15 @@ import org.apache.hyracks.dataflow.std.misc.ReplicateOperatorDescriptor;
  */
 public class FeedOperations {
 
+    private static final CompilerProperties compilerProperties = AppContextInfo.INSTANCE.getCompilerProperties();
+
     private FeedOperations() {
     }
 
     private static Pair<JobSpecification, IAdapterFactory> buildFeedIntakeJobSpec(Feed feed,
             MetadataProvider metadataProvider, FeedPolicyAccessor policyAccessor) throws Exception {
         JobSpecification spec = RuntimeUtils.createJobSpecification();
-        spec.setFrameSize(FeedConstants.JobConstants.DEFAULT_FRAME_SIZE);
+        spec.setFrameSize(compilerProperties.getFrameSize());
         IAdapterFactory adapterFactory;
         IOperatorDescriptor feedIngestor;
         AlgebricksPartitionConstraint ingesterPc;

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/246fbde6/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.1.ddl.aql
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.1.ddl.aql b/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.1.ddl.aql
new file mode 100644
index 0000000..4e21323
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.1.ddl.aql
@@ -0,0 +1,58 @@
+/*
+ * 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.
+ */
+/*
+ * Description  : Create a socket feed with a client that pushes
+ * 10 records. The feed is connected to a dataset that is then
+ * queried for the data.
+ * Expected Res : Success
+ * Date         : 16th Mar 2017
+ */
+drop dataverse experiments if exists;
+create dataverse experiments;
+use dataverse experiments;
+
+create type TwitterUserType as closed {
+    screen-name: string,
+    lang: string,
+    friends_count: int32,
+    statuses_count: int32,
+    name: string,
+    followers_count: int32
+}
+
+create type TweetMessageType as closed {
+    tweetid: string,
+    tweetid-copy:string,
+    user: TwitterUserType,
+    sender-location: point,
+    send-time: datetime,
+    send-time-copy:datetime,
+    referred-topics: {{ string }},
+    message-text: string
+}
+
+create dataset Tweets(TweetMessageType) primary key tweetid;
+
+create feed TweetFeed using socket_adapter
+(
+    ("sockets"="127.0.0.1:10001"),
+    ("address-type"="IP"),
+    ("type-name"="TweetMessageType"),
+    ("format"="adm")
+);
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/246fbde6/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.2.update.aql
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.2.update.aql b/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.2.update.aql
new file mode 100644
index 0000000..40ebc90
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.2.update.aql
@@ -0,0 +1,32 @@
+/*
+ * 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.
+ */
+/*
+ * Description  : Create a socket feed with a client that pushes
+ * 10 records. The feed is connected to a dataset that is then
+ * queried for the data.
+ * Expected Res : Success
+ * Date         : 16th Mar 2017
+ */
+
+use dataverse experiments;
+set wait-for-completion-feed "false";
+
+connect feed TweetFeed to dataset Tweets using policy Discard;
+
+start feed TweetFeed;

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/246fbde6/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.3.server.aql
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.3.server.aql b/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.3.server.aql
new file mode 100644
index 0000000..c78da31
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.3.server.aql
@@ -0,0 +1,26 @@
+/*
+ * 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.
+ */
+/*
+ * Description  : Create a socket feed with a client that pushes
+ * 10 records. The feed is connected to a dataset that is then
+ * queried for the data.
+ * Expected Res : Success
+ * Date         : 16th Mar 2017
+ */
+start client 10001 file-client 127.0.0.1 ../asterix-app/data/twitter/tw_messages.adm 500 50 1000
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/246fbde6/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.4.sleep.aql
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.4.sleep.aql b/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.4.sleep.aql
new file mode 100644
index 0000000..7c1ab60
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.4.sleep.aql
@@ -0,0 +1,26 @@
+/*
+ * 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.
+ */
+/*
+ * Description  : Create a socket feed with a client that pushes
+ * 10 records. The feed is connected to a dataset that is then
+ * queried for the data.
+ * Expected Res : Success
+ * Date         : 16th Mar 2017
+ */
+5000
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/246fbde6/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.5.update.aql
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.5.update.aql b/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.5.update.aql
new file mode 100644
index 0000000..167c507
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.5.update.aql
@@ -0,0 +1,29 @@
+/*
+ * 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.
+ */
+/*
+ * Description  : Create a socket feed with a client that pushes
+ * 10 records. The feed is connected to a dataset that is then
+ * queried for the data.
+ * Expected Res : Success
+ * Date         : 16th Mar 2017
+ */
+
+use dataverse experiments;
+stop feed TweetFeed;
+disconnect feed TweetFeed from dataset Tweets;
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/246fbde6/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.6.query.aql
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.6.query.aql b/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.6.query.aql
new file mode 100644
index 0000000..b67042d
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.6.query.aql
@@ -0,0 +1,31 @@
+/*
+ * 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.
+ */
+/*
+ * Description  : Create a socket feed with a client that pushes
+ * 10 records. The feed is connected to a dataset that is then
+ * queried for the data.
+ * Expected Res : Success
+ * Date         : 16th Mar 2017
+ */
+
+use dataverse experiments;
+
+for $x in dataset Tweets
+order by $x.tweetid
+return $x;
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/246fbde6/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.7.server.aql
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.7.server.aql b/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.7.server.aql
new file mode 100644
index 0000000..cc02ce6
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.7.server.aql
@@ -0,0 +1,27 @@
+/*
+ * 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.
+ */
+/*
+ * Description  : Create a socket feed with a client that pushes
+ * 10 records. The feed is connected to a dataset that is then
+ * queried for the data.
+ * Expected Res : Success
+ * Date         : 16th Mar 2017
+ */
+
+stop 10001
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/246fbde6/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.8.ddl.aql
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.8.ddl.aql b/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.8.ddl.aql
new file mode 100644
index 0000000..406fbbb
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries/feeds/create-feed-with-policy/create-feed-with-policy.8.ddl.aql
@@ -0,0 +1,28 @@
+/*
+ * 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.
+ */
+/*
+ * Description  : Create a socket feed with a client that pushes
+ * 10 records. The feed is connected to a dataset that is then
+ * queried for the data.
+ * Expected Res : Success
+ * Date         : 16th Mar 2017
+ */
+
+use dataverse experiments;
+drop dataverse experiments;
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/246fbde6/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/feeds_08/feeds_08.2.update.sqlpp
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/feeds_08/feeds_08.2.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/feeds_08/feeds_08.2.update.sqlpp
index 9f60585..d679188 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/feeds_08/feeds_08.2.update.sqlpp
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/feeds_08/feeds_08.2.update.sqlpp
@@ -21,6 +21,6 @@ use feeds;
 
 set `wait-for-completion-feed` `true`;
 
-connect  feed SyntheticTweetFeed to  dataset SyntheticTweets using policy `BasicFT`;
+connect  feed SyntheticTweetFeed to  dataset SyntheticTweets using policy `Basic`;
 
 start feed SyntheticTweetFeed;
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/246fbde6/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/feeds_09/feeds_09.2.update.sqlpp
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/feeds_09/feeds_09.2.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/feeds_09/feeds_09.2.update.sqlpp
index 0a940bd..b449744 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/feeds_09/feeds_09.2.update.sqlpp
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/feeds_09/feeds_09.2.update.sqlpp
@@ -19,5 +19,5 @@
 
 use feeds_09;
 set `wait-for-completion-feed` `true`;
-connect  feed SyntheticTweetFeed to  dataset SyntheticTweets using policy `BasicFT`;
+connect  feed SyntheticTweetFeed to  dataset SyntheticTweets using policy `Basic`;
 start feed SyntheticTweetFeed;
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/246fbde6/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/feeds_10/feeds_10.2.update.sqlpp
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/feeds_10/feeds_10.2.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/feeds_10/feeds_10.2.update.sqlpp
index 4c5cc66..011a986 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/feeds_10/feeds_10.2.update.sqlpp
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/feeds_10/feeds_10.2.update.sqlpp
@@ -19,5 +19,5 @@
 
 use feeds_10;
 set `wait-for-completion-feed` `true`;
-connect  feed TweetFeed to  dataset Tweets using policy `BasicFT`;
+connect  feed TweetFeed to  dataset Tweets using policy `Basic`;
 start feed TweetFeed;
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/246fbde6/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/feeds_11/feeds_11.2.update.sqlpp
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/feeds_11/feeds_11.2.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/feeds_11/feeds_11.2.update.sqlpp
index 7af2dbd..327a222 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/feeds_11/feeds_11.2.update.sqlpp
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/feeds_11/feeds_11.2.update.sqlpp
@@ -19,5 +19,5 @@
 
 use feeds;
 set `wait-for-completion-feed` `true`;
-connect  feed feeds.TweetFeed to  dataset feeds.Tweets using policy `BasicFT`;
+connect  feed feeds.TweetFeed to  dataset feeds.Tweets using policy `Basic`;
 start feed feeds.TweetFeed;
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/246fbde6/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/feeds_12/feeds_12.2.update.sqlpp
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/feeds_12/feeds_12.2.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/feeds_12/feeds_12.2.update.sqlpp
index a3f0923..f5d9649 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/feeds_12/feeds_12.2.update.sqlpp
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/feeds_12/feeds_12.2.update.sqlpp
@@ -19,5 +19,5 @@
 
 use feeds_12;
 set `wait-for-completion-feed` `true`;
-connect  feed TweetFeed to  dataset Tweets using policy `BasicFT`;
+connect  feed TweetFeed to  dataset Tweets using policy `Basic`;
 start feed TweetFeed;
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/246fbde6/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/issue_230_feeds/issue_230_feeds.2.update.sqlpp
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/issue_230_feeds/issue_230_feeds.2.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/issue_230_feeds/issue_230_feeds.2.update.sqlpp
index 7af2dbd..327a222 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/issue_230_feeds/issue_230_feeds.2.update.sqlpp
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/feeds/issue_230_feeds/issue_230_feeds.2.update.sqlpp
@@ -19,5 +19,5 @@
 
 use feeds;
 set `wait-for-completion-feed` `true`;
-connect  feed feeds.TweetFeed to  dataset feeds.Tweets using policy `BasicFT`;
+connect  feed feeds.TweetFeed to  dataset feeds.Tweets using policy `Basic`;
 start feed feeds.TweetFeed;
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/246fbde6/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/hints/issue_251_dataset_hint_7/issue_251_dataset_hint_7.2.update.sqlpp
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/hints/issue_251_dataset_hint_7/issue_251_dataset_hint_7.2.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/hints/issue_251_dataset_hint_7/issue_251_dataset_hint_7.2.update.sqlpp
index c5f0625..4915658 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/hints/issue_251_dataset_hint_7/issue_251_dataset_hint_7.2.update.sqlpp
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/hints/issue_251_dataset_hint_7/issue_251_dataset_hint_7.2.update.sqlpp
@@ -29,6 +29,6 @@ use feeds;
 
 set `wait-for-completion-feed` `true`;
 
-connect  feed TweetFeed to  dataset Tweets using policy `BasicFT`;
+connect  feed TweetFeed to  dataset Tweets using policy `Basic`;
 
 start feed TweetFeed;

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/246fbde6/asterixdb/asterix-app/src/test/resources/runtimets/results/feeds/create-feed-with-policy/create-feed-with-policy.1.adm
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/feeds/create-feed-with-policy/create-feed-with-policy.1.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/feeds/create-feed-with-policy/create-feed-with-policy.1.adm
new file mode 100644
index 0000000..6466feb
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/feeds/create-feed-with-policy/create-feed-with-policy.1.adm
@@ -0,0 +1,10 @@
+{ "tweetid": "1", "tweetid-copy": "1", "user": { "screen-name": "RollandEckhardstein#211", "lang": "en", "friends_count": 3657079, "statuses_count": 268, "name": "Rolland Eckhardstein", "followers_count": 3311368 }, "sender-location": point("42.13,80.43"), "send-time": datetime("2005-12-05T21:06:41.000Z"), "send-time-copy": datetime("2005-12-05T21:06:41.000Z"), "referred-topics": {{ "samsung", "plan" }}, "message-text": " love samsung the plan is amazing" }
+{ "tweetid": "10", "tweetid-copy": "10", "user": { "screen-name": "Rolldstein#211", "lang": "en", "friends_count": 3657079, "statuses_count": 268, "name": "Rolland Eckhardstful", "followers_count": 3311368 }, "sender-location": point("46.94,93.98"), "send-time": datetime("2011-04-07T14:08:46.000Z"), "send-time-copy": datetime("2011-04-07T14:08:46.000Z"), "referred-topics": {{ "t-mobile", "signal" }}, "message-text": " like t-mobile the signal is good" }
+{ "tweetid": "2", "tweetid-copy": "2", "user": { "screen-name": "RollandEckhardstein#211", "lang": "en", "friends_count": 3657079, "statuses_count": 268, "name": "David Eckhardstein", "followers_count": 3311368 }, "sender-location": point("28.86,70.44"), "send-time": datetime("2007-08-15T06:44:17.000Z"), "send-time-copy": datetime("2007-08-15T06:44:17.000Z"), "referred-topics": {{ "sprint", "voice-clarity" }}, "message-text": " like sprint its voice-clarity is mind-blowing" }
+{ "tweetid": "3", "tweetid-copy": "3", "user": { "screen-name": "RollandEckhard#500", "lang": "en", "friends_count": 3657079, "statuses_count": 268, "name": "Rolland Hetfield", "followers_count": 3311368 }, "sender-location": point("39.84,86.48"), "send-time": datetime("2008-12-24T00:07:04.000Z"), "send-time-copy": datetime("2008-12-24T00:07:04.000Z"), "referred-topics": {{ "verizon", "voice-command" }}, "message-text": " can't stand verizon its voice-command is terrible:(" }
+{ "tweetid": "4", "tweetid-copy": "4", "user": { "screen-name": "RollandEckhardstein#221", "lang": "en", "friends_count": 3657079, "statuses_count": 268, "name": "Rolland Eckhardstinz", "followers_count": 3311368 }, "sender-location": point("27.67,87.32"), "send-time": datetime("2007-02-05T16:39:13.000Z"), "send-time-copy": datetime("2007-02-05T16:39:13.000Z"), "referred-topics": {{ "t-mobile", "customer-service" }}, "message-text": " love t-mobile its customer-service is mind-blowing" }
+{ "tweetid": "5", "tweetid-copy": "5", "user": { "screen-name": "RollandEcstein#211", "lang": "en", "friends_count": 3657079, "statuses_count": 268, "name": "Rolland Eckhardst", "followers_count": 3311368 }, "sender-location": point("27.3,92.77"), "send-time": datetime("2010-09-12T06:15:28.000Z"), "send-time-copy": datetime("2010-09-12T06:15:28.000Z"), "referred-topics": {{ "t-mobile", "customization" }}, "message-text": " like t-mobile the customization is amazing:)" }
+{ "tweetid": "6", "tweetid-copy": "6", "user": { "screen-name": "Rollkhardstein#211", "lang": "en", "friends_count": 3657079, "statuses_count": 268, "name": "Kirk Hammette ", "followers_count": 3311368 }, "sender-location": point("45.62,84.78"), "send-time": datetime("2012-01-23T06:23:13.000Z"), "send-time-copy": datetime("2012-01-23T06:23:13.000Z"), "referred-topics": {{ "iphone", "network" }}, "message-text": " like iphone its network is awesome:)" }
+{ "tweetid": "7", "tweetid-copy": "7", "user": { "screen-name": "andEckhardstein#211", "lang": "en", "friends_count": 3657079, "statuses_count": 268, "name": "Rolland khardstein", "followers_count": 3311368 }, "sender-location": point("44.12,81.46"), "send-time": datetime("2012-02-17T17:30:26.000Z"), "send-time-copy": datetime("2012-02-17T17:30:26.000Z"), "referred-topics": {{ "t-mobile", "network" }}, "message-text": " hate t-mobile the network is bad" }
+{ "tweetid": "8", "tweetid-copy": "8", "user": { "screen-name": "Rolltein#211", "lang": "en", "friends_count": 3657079, "statuses_count": 268, "name": "Ron Eckhardstein", "followers_count": 3311368 }, "sender-location": point("36.86,90.71"), "send-time": datetime("2009-03-12T13:18:04.000Z"), "send-time-copy": datetime("2009-03-12T13:18:04.000Z"), "referred-topics": {{ "at&t", "touch-screen" }}, "message-text": " dislike at&t its touch-screen is OMG" }
+{ "tweetid": "9", "tweetid-copy": "9", "user": { "screen-name": "Roldstein#211", "lang": "en", "friends_count": 3657079, "statuses_count": 268, "name": "Rolland Eckdstein", "followers_count": 3311368 }, "sender-location": point("29.07,97.05"), "send-time": datetime("2012-08-15T20:19:46.000Z"), "send-time-copy": datetime("2012-08-15T20:19:46.000Z"), "referred-topics": {{ "verizon", "speed" }}, "message-text": " hate verizon its speed is bad" }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/246fbde6/asterixdb/asterix-app/src/test/resources/runtimets/results/feeds/feeds_03/feeds_03.1.adm
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/feeds/feeds_03/feeds_03.1.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/feeds/feeds_03/feeds_03.1.adm
index 573793f..cbde66d 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/results/feeds/feeds_03/feeds_03.1.adm
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/feeds/feeds_03/feeds_03.1.adm
@@ -1 +1 @@
-{ "DataverseName": "feeds", "FeedName": "TweetFeed", "DatasetName": "Tweets", "ReturnType": "TweetType: closed {\n  id: string,\n  username: string,\n  location: string,\n  text: string,\n  timestamp: string\n}\n", "AppliedFunctions": {{ "feed_processor" }}, "PolicyName": "BasicFT" }
+{ "DataverseName": "feeds", "FeedName": "TweetFeed", "DatasetName": "Tweets", "ReturnType": "TweetType: closed {\n  id: string,\n  username: string,\n  location: string,\n  text: string,\n  timestamp: string\n}\n", "AppliedFunctions": {{ "feed_processor" }}, "PolicyName": "Basic" }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/246fbde6/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/dataflow/FeedRuntimeInputHandler.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/dataflow/FeedRuntimeInputHandler.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/dataflow/FeedRuntimeInputHandler.java
index 9982477..329451d 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/dataflow/FeedRuntimeInputHandler.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/dataflow/FeedRuntimeInputHandler.java
@@ -19,7 +19,7 @@
 package org.apache.asterix.external.feed.dataflow;
 
 import java.nio.ByteBuffer;
-import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -59,7 +59,7 @@ public class FeedRuntimeInputHandler extends AbstractUnaryInputUnaryOutputOperat
     private final int initialFrameSize;
     private final FrameTransporter consumer;
     private final Thread consumerThread;
-    private final LinkedBlockingDeque<ByteBuffer> inbox;
+    private final LinkedBlockingQueue<ByteBuffer> inbox;
     private final ConcurrentFramePool framePool;
     private Mode mode = Mode.PROCESS;
     private int total = 0;
@@ -72,7 +72,6 @@ public class FeedRuntimeInputHandler extends AbstractUnaryInputUnaryOutputOperat
             IFrameWriter writer, FeedPolicyAccessor fpa, FrameTupleAccessor fta, ConcurrentFramePool framePool)
             throws HyracksDataException {
         this.writer = writer;
-
         this.spiller = fpa.spillToDiskOnCongestion()
                 ? new FrameSpiller(ctx,
                         connectionId.getFeedId() + "_" + connectionId.getDatasetName() + "_"
@@ -82,10 +81,9 @@ public class FeedRuntimeInputHandler extends AbstractUnaryInputUnaryOutputOperat
         this.exceptionHandler = new FeedExceptionHandler(ctx, fta);
         this.fpa = fpa;
         this.framePool = framePool;
-        this.inbox = new LinkedBlockingDeque<>();
+        this.inbox = new LinkedBlockingQueue<>();
         this.consumer = new FrameTransporter();
-        this.consumerThread = new Thread(consumer);
-        this.consumerThread.start();
+        this.consumerThread = new Thread(consumer, "FeedRuntimeInputHandler-FrameTransporter");
         this.initialFrameSize = ctx.getInitialFrameSize();
         this.frameAction = new FrameAction();
     }
@@ -94,6 +92,7 @@ public class FeedRuntimeInputHandler extends AbstractUnaryInputUnaryOutputOperat
     public void open() throws HyracksDataException {
         synchronized (writer) {
             writer.open();
+            consumerThread.start();
         }
     }
 
@@ -106,14 +105,18 @@ public class FeedRuntimeInputHandler extends AbstractUnaryInputUnaryOutputOperat
 
     @Override
     public void close() throws HyracksDataException {
-        consumer.poison();
-        synchronized (mutex) {
-            if (DEBUG) {
-                LOGGER.info("Producer is waking up consumer");
-            }
-            mutex.notify();
-        }
         try {
+            // Here we only put the poison frame into the inbox.
+            // If we use nextframe, chances are this frame will also be
+            // flushed into the spilled file. This causes problem when trying to
+            // read the frame and the size info is lost.
+            inbox.put(ByteBuffer.allocate(0));
+            synchronized (mutex) {
+                if (DEBUG) {
+                    LOGGER.info("Producer is waking up consumer");
+                }
+                mutex.notify();
+            }
             consumerThread.join();
         } catch (InterruptedException e) {
             LOGGER.log(Level.WARNING, e.getMessage(), e);
@@ -391,9 +394,7 @@ public class FeedRuntimeInputHandler extends AbstractUnaryInputUnaryOutputOperat
 
     @Override
     public void flush() throws HyracksDataException {
-        synchronized (writer) {
-            writer.flush();
-        }
+        // no op
     }
 
     public int getNumDiscarded() {
@@ -415,16 +416,11 @@ public class FeedRuntimeInputHandler extends AbstractUnaryInputUnaryOutputOperat
     private class FrameTransporter implements Runnable {
         private volatile Throwable cause;
         private int consumed = 0;
-        private boolean poisoned = false;
 
         public Throwable cause() {
             return cause;
         }
 
-        public void poison() {
-            poisoned = true;
-        }
-
         private Throwable consume(ByteBuffer frame) {
             while (frame != null) {
                 try {
@@ -446,61 +442,59 @@ public class FeedRuntimeInputHandler extends AbstractUnaryInputUnaryOutputOperat
             return null;
         }
 
+        private boolean clearLocalFrames() throws HyracksDataException {
+            ByteBuffer frame = spiller.next();
+            while (frame != null) {
+                if (consume(frame) != null) {
+                    return false;
+                }
+                frame = spiller.next();
+            }
+            return true;
+        }
+
         @Override
         public void run() {
             try {
-                ByteBuffer frame = inbox.poll();
-                while (true) {
-                    if (frame != null) {
+                ByteBuffer frame;
+                boolean running = true;
+                while (running) {
+                    frame = inbox.poll();
+
+                    if (frame == null && spiller != null) {
+                        running = clearLocalFrames();
+                        continue;
+                    }
+
+                    if (frame == null) {
+                        synchronized (mutex) {
+                            LOGGER.info("Consumer is going to sleep");
+                            mutex.wait();
+                            LOGGER.info("Consumer is waking up");
+                        }
+                        continue;
+                    }
+
+                    // process
+                    if (frame.capacity() == 0) {
+                        running = false;
+                        if (spiller != null ) {
+                            clearLocalFrames();
+                        }
+                    } else {
                         try {
                             if (consume(frame) != null) {
                                 return;
                             }
                         } finally {
-                            // Done with frame.
                             framePool.release(frame);
                         }
                     }
-                    frame = inbox.poll();
-                    if (frame == null) {
-                        // Memory queue is empty. Check spill
-                        if (spiller != null) {
-                            frame = spiller.next();
-                            while (frame != null) {
-                                if (consume(frame) != null) {
-                                    // We don't release the frame since this is a spill frame that we didn't get from memory
-                                    // manager
-                                    return;
-                                }
-                                frame = spiller.next();
-                            }
-                        }
-                        writer.flush();
-                        // At this point. We consumed all memory and spilled
-                        // We can't assume the next will be in memory. what if there is 0 memory?
-                        synchronized (mutex) {
-                            frame = inbox.poll();
-                            // Nothing in memory
-                            if (frame == null && (spiller == null || spiller.switchToMemory())) {
-                                if (poisoned) {
-                                    break;
-                                }
-                                if (DEBUG) {
-                                    LOGGER.info("Consumer is going to sleep");
-                                }
-                                // Nothing in disk
-                                mutex.wait();
-                                if (DEBUG) {
-                                    LOGGER.info("Consumer is waking up");
-                                }
-                            }
-                        }
-                    }
+                    writer.flush();
                 }
             } catch (Throwable th) {
                 this.cause = th;
             }
-            // cleanup will always be done through the close() call
         }
 
         @Override
@@ -513,7 +507,7 @@ public class FeedRuntimeInputHandler extends AbstractUnaryInputUnaryOutputOperat
         return total;
     }
 
-    public LinkedBlockingDeque<ByteBuffer> getInternalBuffer() {
+    public LinkedBlockingQueue<ByteBuffer> getInternalBuffer() {
         return inbox;
     }
 }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/246fbde6/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/policy/FeedPolicyAccessor.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/policy/FeedPolicyAccessor.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/policy/FeedPolicyAccessor.java
index cb722bb..becf41d 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/policy/FeedPolicyAccessor.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/policy/FeedPolicyAccessor.java
@@ -30,33 +30,12 @@ public class FeedPolicyAccessor implements Serializable {
 
     /**
      * --------------------------
-     * failure configuration
-     * --------------------------
-     **/
-
-    /** continue feed ingestion after a soft (runtime) failure **/
-    public static final String SOFT_FAILURE_CONTINUE = "soft.failure.continue";
-
-    /** log failed tuple to an asterixdb dataset for future reference **/
-    public static final String SOFT_FAILURE_LOG_DATA = "soft.failure.log.data";
-
-    /** continue feed ingestion after loss of one or more machines (hardware failure) **/
-    public static final String HARDWARE_FAILURE_CONTINUE = "hardware.failure.continue";
-
-    /** auto-start a loser feed when the asterixdb instance is restarted **/
-    public static final String CLUSTER_REBOOT_AUTO_RESTART = "cluster.reboot.auto.restart";
-
-    /** framework provides guarantee that each received feed record will be processed through the ingestion pipeline at least once **/
-    public static final String AT_LEAST_ONE_SEMANTICS = "atleast.once.semantics";
-
-    /**
-     * --------------------------
      * flow control configuration
      * --------------------------
      **/
 
     /** enable buffering in feeds **/
-    public static final String BUFFERING_ENABLED = "buffering.enabled";
+    public static final String FLOWCONTROL_ENABLED = "flowcontrol.enabled";
 
     /** spill excess tuples to disk if an operator cannot process incoming data at its arrival rate **/
     public static final String SPILL_TO_DISK_ON_CONGESTION = "spill.to.disk.on.congestion";
@@ -70,18 +49,9 @@ public class FeedPolicyAccessor implements Serializable {
     /** maximum fraction of ingested data that can be discarded **/
     public static final String MAX_FRACTION_DISCARD = "max.fraction.discard";
 
-    /** maximum end-to-end delay/latency in persisting a tuple through the feed ingestion pipeline **/
-    public static final String MAX_DELAY_RECORD_PERSISTENCE = "max.delay.record.persistence";
-
-    /** rate limit the inflow of tuples in accordance with the maximum capacity of the pipeline **/
-    public static final String THROTTLING_ENABLED = "throttling.enabled";
-
     /** elasticity **/
     public static final String ELASTIC = "elastic";
 
-    /** statistics **/
-    public static final String TIME_TRACKING = "time.tracking";
-
     /** logging of statistics **/
     public static final String LOGGING_STATISTICS = "logging.statistics";
 
@@ -101,31 +71,9 @@ public class FeedPolicyAccessor implements Serializable {
         this.feedPolicy = feedPolicy;
     }
 
-    /** Failure recover/reporting **/
-
-    public boolean logDataOnSoftFailure() {
-        return getBooleanPropertyValue(SOFT_FAILURE_LOG_DATA, false);
-    }
-
-    public boolean continueOnSoftFailure() {
-        return getBooleanPropertyValue(SOFT_FAILURE_CONTINUE, false);
-    }
-
-    public boolean continueOnHardwareFailure() {
-        return getBooleanPropertyValue(HARDWARE_FAILURE_CONTINUE, false);
-    }
-
-    public boolean autoRestartOnClusterReboot() {
-        return getBooleanPropertyValue(CLUSTER_REBOOT_AUTO_RESTART, false);
-    }
-
-    public boolean atleastOnceSemantics() {
-        return getBooleanPropertyValue(AT_LEAST_ONE_SEMANTICS, false);
-    }
-
     /** flow control **/
-    public boolean bufferingEnabled() {
-        return getBooleanPropertyValue(BUFFERING_ENABLED, false);
+    public boolean flowControlEnabled() {
+        return getBooleanPropertyValue(FLOWCONTROL_ENABLED, false);
     }
 
     public boolean spillToDiskOnCongestion() {
@@ -136,10 +84,6 @@ public class FeedPolicyAccessor implements Serializable {
         return getMaxFractionDiscard() > 0;
     }
 
-    public boolean throttlingEnabled() {
-        return getBooleanPropertyValue(THROTTLING_ENABLED, false);
-    }
-
     public long getMaxSpillOnDisk() {
         return getLongPropertyValue(MAX_SPILL_SIZE_ON_DISK, NO_LIMIT);
     }
@@ -148,28 +92,9 @@ public class FeedPolicyAccessor implements Serializable {
         return getFloatPropertyValue(MAX_FRACTION_DISCARD, 0);
     }
 
-    public long getMaxDelayRecordPersistence() {
-        return getLongPropertyValue(MAX_DELAY_RECORD_PERSISTENCE, Long.MAX_VALUE);
-    }
-
-    /** Elasticity **/
-    public boolean isElastic() {
-        return getBooleanPropertyValue(ELASTIC, false);
-    }
-
-    /** Statistics **/
-    public boolean isTimeTrackingEnabled() {
-        return getBooleanPropertyValue(TIME_TRACKING, false);
-    }
-
-    /** Logging of statistics **/
-    public boolean isLoggingStatisticsEnabled() {
-        return getBooleanPropertyValue(LOGGING_STATISTICS, false);
-    }
-
     private boolean getBooleanPropertyValue(String key, boolean defValue) {
         String v = feedPolicy.get(key);
-        return v == null ? false : Boolean.valueOf(v);
+        return v == null ? defValue : Boolean.valueOf(v);
     }
 
     private long getLongPropertyValue(String key, long defValue) {

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/246fbde6/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/policy/FeedPolicyEnforcer.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/policy/FeedPolicyEnforcer.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/policy/FeedPolicyEnforcer.java
deleted file mode 100644
index 3483da4..0000000
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/policy/FeedPolicyEnforcer.java
+++ /dev/null
@@ -1,48 +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.asterix.external.feed.policy;
-
-import java.util.Map;
-
-import org.apache.asterix.external.feed.management.FeedConnectionId;
-import org.apache.hyracks.api.exceptions.HyracksDataException;
-
-public class FeedPolicyEnforcer {
-
-    private final FeedConnectionId connectionId;
-    private final FeedPolicyAccessor policyAccessor;
-
-    public FeedPolicyEnforcer(FeedConnectionId feedConnectionId, Map<String, String> feedPolicy) {
-        this.connectionId = feedConnectionId;
-        this.policyAccessor = new FeedPolicyAccessor(feedPolicy);
-    }
-
-    public boolean continueIngestionPostSoftwareFailure(HyracksDataException e) throws HyracksDataException {
-        return policyAccessor.continueOnSoftFailure();
-    }
-
-    public FeedPolicyAccessor getFeedPolicyAccessor() {
-        return policyAccessor;
-    }
-
-    public FeedConnectionId getFeedId() {
-        return connectionId;
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/246fbde6/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/SocketServerInputStream.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/SocketServerInputStream.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/SocketServerInputStream.java
index 964508f..ea55b5b 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/SocketServerInputStream.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/SocketServerInputStream.java
@@ -71,7 +71,8 @@ public class SocketServerInputStream extends AsterixInputStream {
             }
             read = connectionStream.read(b, off, len);
         } catch (IOException e) {
-            e.printStackTrace();
+            // exception is expected when no connection available
+            LOGGER.info("Exhausted all pending connections. Waiting for new ones to come.");
             read = -1;
         }
         while (read < 0) {
@@ -155,11 +156,10 @@ public class SocketServerInputStream extends AsterixInputStream {
     @Override
     public boolean handleException(Throwable th) {
         try {
-            accept();
+            return accept();
         } catch (IOException e) {
             LOGGER.warn("Failed accepting more connections", e);
             return false;
         }
-        return true;
     }
 }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/246fbde6/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedCollectOperatorNodePushable.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedCollectOperatorNodePushable.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedCollectOperatorNodePushable.java
index bdc11f5..f289361 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedCollectOperatorNodePushable.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedCollectOperatorNodePushable.java
@@ -61,7 +61,7 @@ public class FeedCollectOperatorNodePushable extends AbstractUnaryInputUnaryOutp
             ActiveRuntimeId runtimeId =
                     new ActiveRuntimeId(connectionId.getFeedId(), FeedRuntimeType.COLLECT.toString(), partition);
             FrameTupleAccessor tAccessor = new FrameTupleAccessor(recordDesc);
-            if (policyAccessor.bufferingEnabled()) {
+            if (policyAccessor.flowControlEnabled()) {
                 writer = new FeedRuntimeInputHandler(ctx, connectionId, runtimeId, writer, policyAccessor, tAccessor,
                         activeManager.getFramePool());
             } else {

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/246fbde6/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedMetaComputeNodePushable.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedMetaComputeNodePushable.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedMetaComputeNodePushable.java
index fbdbece..6732b15 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedMetaComputeNodePushable.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedMetaComputeNodePushable.java
@@ -30,7 +30,6 @@ import org.apache.asterix.external.feed.dataflow.FeedRuntimeInputHandler;
 import org.apache.asterix.external.feed.dataflow.SyncFeedRuntimeInputHandler;
 import org.apache.asterix.external.feed.management.FeedConnectionId;
 import org.apache.asterix.external.feed.policy.FeedPolicyAccessor;
-import org.apache.asterix.external.feed.policy.FeedPolicyEnforcer;
 import org.apache.asterix.external.util.FeedUtils;
 import org.apache.asterix.external.util.FeedUtils.FeedRuntimeType;
 import org.apache.hyracks.api.comm.VSizeFrame;
@@ -55,10 +54,10 @@ public class FeedMetaComputeNodePushable extends AbstractUnaryInputUnaryOutputOp
     private AbstractUnaryInputUnaryOutputOperatorNodePushable coreOperator;
 
     /**
-     * A policy enforcer that ensures dynamic decisions for a feed are taken
+     * A policy accessor that ensures dynamic decisions for a feed are taken
      * in accordance with the associated ingestion policy
      **/
-    private FeedPolicyEnforcer policyEnforcer;
+    private FeedPolicyAccessor policyAccessor;
 
     /**
      * A unique identifier for the feed instance. A feed instance represents
@@ -101,7 +100,7 @@ public class FeedMetaComputeNodePushable extends AbstractUnaryInputUnaryOutputOp
         this.ctx = ctx;
         this.coreOperator = (AbstractUnaryInputUnaryOutputOperatorNodePushable) ((IActivity) coreOperator)
                 .createPushRuntime(ctx, recordDescProvider, partition, nPartitions);
-        this.policyEnforcer = new FeedPolicyEnforcer(feedConnectionId, feedPolicyProperties);
+        this.policyAccessor = new FeedPolicyAccessor(feedPolicyProperties);
         this.partition = partition;
         this.connectionId = feedConnectionId;
         this.feedManager = (ActiveManager) ((IAppRuntimeContext) ctx.getJobletContext().getServiceContext()
@@ -127,9 +126,9 @@ public class FeedMetaComputeNodePushable extends AbstractUnaryInputUnaryOutputOp
 
     private void initializeNewFeedRuntime(ActiveRuntimeId runtimeId) throws Exception {
         fta = new FrameTupleAccessor(recordDescProvider.getInputRecordDescriptor(opDesc.getActivityId(), 0));
-        FeedPolicyAccessor fpa = policyEnforcer.getFeedPolicyAccessor();
+        FeedPolicyAccessor fpa = policyAccessor;
         coreOperator.setOutputFrameWriter(0, writer, recordDesc);
-        if (fpa.bufferingEnabled()) {
+        if (fpa.flowControlEnabled()) {
             writer = new FeedRuntimeInputHandler(ctx, connectionId, runtimeId, coreOperator, fpa, fta,
                     feedManager.getFramePool());
         } else {

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/246fbde6/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedMetaStoreNodePushable.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedMetaStoreNodePushable.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedMetaStoreNodePushable.java
index 0bb27db..f2193af 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedMetaStoreNodePushable.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedMetaStoreNodePushable.java
@@ -30,7 +30,7 @@ import org.apache.asterix.common.dataflow.LSMInsertDeleteOperatorNodePushable;
 import org.apache.asterix.external.feed.dataflow.FeedRuntimeInputHandler;
 import org.apache.asterix.external.feed.dataflow.SyncFeedRuntimeInputHandler;
 import org.apache.asterix.external.feed.management.FeedConnectionId;
-import org.apache.asterix.external.feed.policy.FeedPolicyEnforcer;
+import org.apache.asterix.external.feed.policy.FeedPolicyAccessor;
 import org.apache.asterix.external.util.FeedUtils;
 import org.apache.asterix.external.util.FeedUtils.FeedRuntimeType;
 import org.apache.hyracks.api.comm.VSizeFrame;
@@ -52,10 +52,10 @@ public class FeedMetaStoreNodePushable extends AbstractUnaryInputUnaryOutputOper
     private AbstractUnaryInputUnaryOutputOperatorNodePushable insertOperator;
 
     /**
-     * A policy enforcer that ensures dyanmic decisions for a feed are taken
+     * A policy accessor that ensures dyanmic decisions for a feed are taken
      * in accordance with the associated ingestion policy
      **/
-    private final FeedPolicyEnforcer policyEnforcer;
+    private final FeedPolicyAccessor policyAccessor;
 
     /**
      * A unique identifier for the feed instance. A feed instance represents
@@ -94,7 +94,7 @@ public class FeedMetaStoreNodePushable extends AbstractUnaryInputUnaryOutputOper
         this.ctx = ctx;
         this.insertOperator = (AbstractUnaryInputUnaryOutputOperatorNodePushable) ((IActivity) coreOperator)
                 .createPushRuntime(ctx, recordDescProvider, partition, nPartitions);
-        this.policyEnforcer = new FeedPolicyEnforcer(feedConnectionId, feedPolicyProperties);
+        this.policyAccessor = new FeedPolicyAccessor(feedPolicyProperties);
         this.partition = partition;
         this.connectionId = feedConnectionId;
         this.feedManager = (ActiveManager) ((IAppRuntimeContext) ctx.getJobletContext().getServiceContext()
@@ -130,9 +130,9 @@ public class FeedMetaStoreNodePushable extends AbstractUnaryInputUnaryOutputOper
                 return;
             }
         }
-        if (policyEnforcer.getFeedPolicyAccessor().bufferingEnabled()) {
+        if (policyAccessor.flowControlEnabled()) {
             writer = new FeedRuntimeInputHandler(ctx, connectionId, runtimeId, insertOperator,
-                    policyEnforcer.getFeedPolicyAccessor(), fta, feedManager.getFramePool());
+                    policyAccessor, fta, feedManager.getFramePool());
         } else {
             writer = new SyncFeedRuntimeInputHandler(ctx, insertOperator, fta);
         }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/246fbde6/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/FeedConstants.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/FeedConstants.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/FeedConstants.java
index cc21360..9538711 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/FeedConstants.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/FeedConstants.java
@@ -69,8 +69,4 @@ public class FeedConstants {
     public static final class NamingConstants {
         public static final String LIBRARY_NAME_SEPARATOR = "#";
     }
-
-    public static class JobConstants {
-        public static final int DEFAULT_FRAME_SIZE = 8192;
-    }
 }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/246fbde6/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/feed/test/InputHandlerTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/feed/test/InputHandlerTest.java b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/feed/test/InputHandlerTest.java
index d407b8a..2237505 100644
--- a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/feed/test/InputHandlerTest.java
+++ b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/feed/test/InputHandlerTest.java
@@ -18,7 +18,11 @@
  */
 package org.apache.asterix.external.feed.test;
 
+import java.io.File;
+import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.Random;
 import java.util.concurrent.ExecutorService;
@@ -33,6 +37,8 @@ import org.apache.asterix.external.feed.management.FeedConnectionId;
 import org.apache.asterix.external.feed.policy.FeedPolicyAccessor;
 import org.apache.asterix.external.util.FeedUtils;
 import org.apache.asterix.external.util.FeedUtils.FeedRuntimeType;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.io.filefilter.WildcardFileFilter;
 import org.apache.hyracks.api.comm.IFrameWriter;
 import org.apache.hyracks.api.comm.VSizeFrame;
 import org.apache.hyracks.api.context.IHyracksTaskContext;
@@ -100,7 +106,7 @@ public class InputHandlerTest extends TestCase {
     private static FeedPolicyAccessor createFeedPolicyAccessor(boolean spill, boolean discard, long spillBudget,
             float discardFraction) {
         FeedPolicyAccessor fpa = Mockito.mock(FeedPolicyAccessor.class);
-        Mockito.when(fpa.bufferingEnabled()).thenReturn(true);
+        Mockito.when(fpa.flowControlEnabled()).thenReturn(true);
         Mockito.when(fpa.spillToDiskOnCongestion()).thenReturn(spill);
         Mockito.when(fpa.getMaxSpillOnDisk()).thenReturn(spillBudget);
         Mockito.when(fpa.discardOnCongestion()).thenReturn(discard);
@@ -108,6 +114,24 @@ public class InputHandlerTest extends TestCase {
         return fpa;
     }
 
+    private void cleanDiskFiles() throws IOException {
+        String filePrefix = "dataverse.feed(Feed)_dataset*";
+        Collection<File> files = FileUtils.listFiles(new File("."), new WildcardFileFilter(filePrefix), null);
+        for (File ifile : files) {
+            Files.deleteIfExists(ifile.toPath());
+        }
+    }
+
+    @org.junit.Before
+    public void testCleanBefore() throws IOException {
+        cleanDiskFiles();
+    }
+
+    @org.junit.After
+    public void testCleanAfter() throws IOException {
+        cleanDiskFiles();
+    }
+
     @org.junit.Test
     public void testZeroMemoryVarSizeFrameWithDiskNoDiscard() {
         try {

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/246fbde6/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/feeds/BuiltinFeedPolicies.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/feeds/BuiltinFeedPolicies.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/feeds/BuiltinFeedPolicies.java
index 57284b0..dfa00ab 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/feeds/BuiltinFeedPolicies.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/feeds/BuiltinFeedPolicies.java
@@ -27,26 +27,18 @@ import org.apache.asterix.metadata.utils.MetadataConstants;
 
 public class BuiltinFeedPolicies {
 
-    public static final FeedPolicyEntity BRITTLE = initializeBrittlePolicy();
-
     public static final FeedPolicyEntity BASIC = initializeBasicPolicy();
 
-    public static final FeedPolicyEntity BASIC_FT = initializeBasicFTPolicy();
-
-    public static final FeedPolicyEntity ADVANCED_FT = initializeAdvancedFTPolicy();
-
     public static final FeedPolicyEntity ADVANCED_FT_DISCARD = initializeAdvancedFTDiscardPolicy();
 
     public static final FeedPolicyEntity ADVANCED_FT_SPILL = initializeAdvancedFTSpillPolicy();
 
-    public static final FeedPolicyEntity ADVANCED_FT_THROTTLE = initializeAdvancedFTThrottlePolicy();
-
     public static final FeedPolicyEntity ELASTIC = initializeAdvancedFTElasticPolicy();
 
-    public static final FeedPolicyEntity[] policies = new FeedPolicyEntity[] { BRITTLE, BASIC, BASIC_FT, ADVANCED_FT,
-            ADVANCED_FT_DISCARD, ADVANCED_FT_SPILL, ADVANCED_FT_THROTTLE, ELASTIC };
+    public static final FeedPolicyEntity[] policies = new FeedPolicyEntity[] { BASIC, ADVANCED_FT_DISCARD,
+            ADVANCED_FT_SPILL, ELASTIC };
 
-    public static final FeedPolicyEntity DEFAULT_POLICY = BASIC_FT;
+    public static final FeedPolicyEntity DEFAULT_POLICY = BASIC;
 
     public static final String CONFIG_FEED_POLICY_KEY = "policy";
 
@@ -59,132 +51,48 @@ public class BuiltinFeedPolicies {
         return null;
     }
 
-    //Brittle
-    private static FeedPolicyEntity initializeBrittlePolicy() {
-        Map<String, String> policyParams = new HashMap<String, String>();
-        policyParams.put(FeedPolicyAccessor.SOFT_FAILURE_CONTINUE, "false");
-        policyParams.put(FeedPolicyAccessor.SOFT_FAILURE_LOG_DATA, "false");
-        policyParams.put(FeedPolicyAccessor.HARDWARE_FAILURE_CONTINUE, "false");
-        policyParams.put(FeedPolicyAccessor.CLUSTER_REBOOT_AUTO_RESTART, "false");
-        policyParams.put(FeedPolicyAccessor.ELASTIC, "false");
-        policyParams.put(FeedPolicyAccessor.TIME_TRACKING, "false");
-        policyParams.put(FeedPolicyAccessor.AT_LEAST_ONE_SEMANTICS, "false");
-
-        String description = "Brittle";
-        return new FeedPolicyEntity(MetadataConstants.METADATA_DATAVERSE_NAME, "Brittle", description, policyParams);
-    }
-
     //Basic
     private static FeedPolicyEntity initializeBasicPolicy() {
         Map<String, String> policyParams = new HashMap<String, String>();
-        policyParams.put(FeedPolicyAccessor.SOFT_FAILURE_CONTINUE, "false");
-        policyParams.put(FeedPolicyAccessor.SOFT_FAILURE_LOG_DATA, "true");
-        policyParams.put(FeedPolicyAccessor.CLUSTER_REBOOT_AUTO_RESTART, "true");
         policyParams.put(FeedPolicyAccessor.ELASTIC, "false");
-        policyParams.put(FeedPolicyAccessor.TIME_TRACKING, "false");
-        policyParams.put(FeedPolicyAccessor.AT_LEAST_ONE_SEMANTICS, "false");
 
         String description = "Basic";
         return new FeedPolicyEntity(MetadataConstants.METADATA_DATAVERSE_NAME, "Basic", description, policyParams);
     }
 
-    // BasicFT
-    private static FeedPolicyEntity initializeBasicFTPolicy() {
-        Map<String, String> policyParams = new HashMap<String, String>();
-        policyParams.put(FeedPolicyAccessor.SOFT_FAILURE_CONTINUE, "true");
-        policyParams.put(FeedPolicyAccessor.SOFT_FAILURE_LOG_DATA, "true");
-        policyParams.put(FeedPolicyAccessor.HARDWARE_FAILURE_CONTINUE, "false");
-        policyParams.put(FeedPolicyAccessor.CLUSTER_REBOOT_AUTO_RESTART, "true");
-        policyParams.put(FeedPolicyAccessor.ELASTIC, "false");
-        policyParams.put(FeedPolicyAccessor.SPILL_TO_DISK_ON_CONGESTION, "false");
-        policyParams.put(FeedPolicyAccessor.MAX_FRACTION_DISCARD, "1");
-        policyParams.put(FeedPolicyAccessor.TIME_TRACKING, "false");
-        policyParams.put(FeedPolicyAccessor.AT_LEAST_ONE_SEMANTICS, "false");
-        policyParams.put(FeedPolicyAccessor.THROTTLING_ENABLED, "false");
-
-        String description = "Basic Monitored Fault-Tolerant";
-        return new FeedPolicyEntity(MetadataConstants.METADATA_DATAVERSE_NAME, "BasicFT", description, policyParams);
-    }
-
-    // AdvancedFT
-    private static FeedPolicyEntity initializeAdvancedFTPolicy() {
-        Map<String, String> policyParams = new HashMap<String, String>();
-        policyParams.put(FeedPolicyAccessor.SOFT_FAILURE_CONTINUE, "true");
-        policyParams.put(FeedPolicyAccessor.SOFT_FAILURE_LOG_DATA, "true");
-        policyParams.put(FeedPolicyAccessor.HARDWARE_FAILURE_CONTINUE, "true");
-        policyParams.put(FeedPolicyAccessor.CLUSTER_REBOOT_AUTO_RESTART, "true");
-        policyParams.put(FeedPolicyAccessor.ELASTIC, "false");
-        policyParams.put(FeedPolicyAccessor.TIME_TRACKING, "true");
-        policyParams.put(FeedPolicyAccessor.AT_LEAST_ONE_SEMANTICS, "true");
-
-        String description = "Basic Monitored Fault-Tolerant with at least once semantics";
-        return new FeedPolicyEntity(MetadataConstants.METADATA_DATAVERSE_NAME, "AdvancedFT", description, policyParams);
-    }
-
-    // AdvancedFT_Discard
+    // Discard
     private static FeedPolicyEntity initializeAdvancedFTDiscardPolicy() {
         Map<String, String> policyParams = new HashMap<String, String>();
-        policyParams.put(FeedPolicyAccessor.SOFT_FAILURE_CONTINUE, "true");
-        policyParams.put(FeedPolicyAccessor.SOFT_FAILURE_LOG_DATA, "true");
-        policyParams.put(FeedPolicyAccessor.HARDWARE_FAILURE_CONTINUE, "true");
-        policyParams.put(FeedPolicyAccessor.CLUSTER_REBOOT_AUTO_RESTART, "true");
         policyParams.put(FeedPolicyAccessor.ELASTIC, "false");
+        policyParams.put(FeedPolicyAccessor.FLOWCONTROL_ENABLED, "true");
         policyParams.put(FeedPolicyAccessor.MAX_SPILL_SIZE_ON_DISK, "false");
         policyParams.put(FeedPolicyAccessor.MAX_FRACTION_DISCARD, "100");
-        policyParams.put(FeedPolicyAccessor.TIME_TRACKING, "false");
         policyParams.put(FeedPolicyAccessor.LOGGING_STATISTICS, "true");
 
-        String description = "AdvancedFT 100% Discard during congestion";
-        return new FeedPolicyEntity(MetadataConstants.METADATA_DATAVERSE_NAME, "AdvancedFT_Discard", description,
+        String description = "FlowControl 100% Discard during congestion";
+        return new FeedPolicyEntity(MetadataConstants.METADATA_DATAVERSE_NAME, "Discard", description,
                 policyParams);
     }
 
-    // AdvancedFT_Spill
+    // Spill
     private static FeedPolicyEntity initializeAdvancedFTSpillPolicy() {
         Map<String, String> policyParams = new HashMap<String, String>();
-        policyParams.put(FeedPolicyAccessor.SOFT_FAILURE_CONTINUE, "true");
-        policyParams.put(FeedPolicyAccessor.SOFT_FAILURE_LOG_DATA, "true");
-        policyParams.put(FeedPolicyAccessor.HARDWARE_FAILURE_CONTINUE, "true");
-        policyParams.put(FeedPolicyAccessor.CLUSTER_REBOOT_AUTO_RESTART, "true");
         policyParams.put(FeedPolicyAccessor.ELASTIC, "false");
+        policyParams.put(FeedPolicyAccessor.FLOWCONTROL_ENABLED, "true");
         policyParams.put(FeedPolicyAccessor.SPILL_TO_DISK_ON_CONGESTION, "" + Boolean.TRUE);
         policyParams.put(FeedPolicyAccessor.MAX_SPILL_SIZE_ON_DISK, "" + FeedPolicyAccessor.NO_LIMIT);
-        policyParams.put(FeedPolicyAccessor.TIME_TRACKING, "true");
-
-        String description = "AdvancedFT 100% Discard during congestion";
-        return new FeedPolicyEntity(MetadataConstants.METADATA_DATAVERSE_NAME, "AdvancedFT_Spill", description,
-                policyParams);
-    }
-
-    // AdvancedFT_Spill
-    private static FeedPolicyEntity initializeAdvancedFTThrottlePolicy() {
-        Map<String, String> policyParams = new HashMap<String, String>();
-        policyParams.put(FeedPolicyAccessor.SOFT_FAILURE_CONTINUE, "true");
-        policyParams.put(FeedPolicyAccessor.SOFT_FAILURE_LOG_DATA, "true");
-        policyParams.put(FeedPolicyAccessor.HARDWARE_FAILURE_CONTINUE, "true");
-        policyParams.put(FeedPolicyAccessor.CLUSTER_REBOOT_AUTO_RESTART, "true");
-        policyParams.put(FeedPolicyAccessor.ELASTIC, "false");
-        policyParams.put(FeedPolicyAccessor.SPILL_TO_DISK_ON_CONGESTION, "" + Boolean.FALSE);
-        policyParams.put(FeedPolicyAccessor.MAX_FRACTION_DISCARD, "" + 0);
-        policyParams.put(FeedPolicyAccessor.TIME_TRACKING, "false");
-        policyParams.put(FeedPolicyAccessor.THROTTLING_ENABLED, "true");
 
-        String description = "AdvancedFT Throttle during congestion";
-        return new FeedPolicyEntity(MetadataConstants.METADATA_DATAVERSE_NAME, "AdvancedFT_Throttle", description,
+        String description = "FlowControl 100% Spill during congestion";
+        return new FeedPolicyEntity(MetadataConstants.METADATA_DATAVERSE_NAME, "Spill", description,
                 policyParams);
     }
 
     // AdvancedFT_Elastic
     private static FeedPolicyEntity initializeAdvancedFTElasticPolicy() {
         Map<String, String> policyParams = new HashMap<String, String>();
-        policyParams.put(FeedPolicyAccessor.SOFT_FAILURE_CONTINUE, "true");
-        policyParams.put(FeedPolicyAccessor.SOFT_FAILURE_LOG_DATA, "true");
-        policyParams.put(FeedPolicyAccessor.HARDWARE_FAILURE_CONTINUE, "true");
-        policyParams.put(FeedPolicyAccessor.CLUSTER_REBOOT_AUTO_RESTART, "true");
         policyParams.put(FeedPolicyAccessor.ELASTIC, "true");
-        policyParams.put(FeedPolicyAccessor.TIME_TRACKING, "false");
+        policyParams.put(FeedPolicyAccessor.FLOWCONTROL_ENABLED, "true");
         policyParams.put(FeedPolicyAccessor.LOGGING_STATISTICS, "true");
-
         String description = "Basic Monitored Fault-Tolerant Elastic";
         return new FeedPolicyEntity(MetadataConstants.METADATA_DATAVERSE_NAME, "AdvancedFT_Elastic", description,
                 policyParams);


Mime
View raw message