You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by am...@apache.org on 2016/01/14 21:32:08 UTC
[14/26] incubator-asterixdb git commit: Feed Fixes and Cleanup
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/e800e6d5/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedConnectionRequest.java
----------------------------------------------------------------------
diff --git a/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedConnectionRequest.java b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedConnectionRequest.java
new file mode 100644
index 0000000..5f22a72
--- /dev/null
+++ b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedConnectionRequest.java
@@ -0,0 +1,125 @@
+/*
+ * 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.management;
+
+import java.util.List;
+import java.util.Map;
+
+import org.apache.asterix.external.feed.api.IFeedLifecycleListener.ConnectionLocation;
+import org.apache.commons.lang3.StringUtils;
+
+/**
+ * A request for connecting a feed to a dataset.
+ */
+public class FeedConnectionRequest {
+
+ public enum ConnectionStatus {
+ /** initial state upon creating a connection request **/
+ INITIALIZED,
+
+ /** connection establish; feed is receiving data **/
+ ACTIVE,
+
+ /** connection removed; feed is not receiving data **/
+ INACTIVE,
+
+ /** connection request failed **/
+ FAILED
+ }
+
+ /** Feed joint on the feed pipeline that serves as the source for this subscription **/
+ private final FeedJointKey feedJointKey;
+
+ /** Location in the source feed pipeline from where feed tuples are received. **/
+ private final ConnectionLocation connectionLocation;
+
+ /** List of functions that need to be applied in sequence after the data hand-off at the source feedPointKey. **/
+ private final List<String> functionsToApply;
+
+ /** Status associated with the subscription. */
+ private ConnectionStatus connectionStatus;
+
+ /** Name of the policy that governs feed ingestion **/
+ private final String policy;
+
+ /** Policy associated with a feed connection **/
+ private final Map<String, String> policyParameters;
+
+ /** Target dataset associated with the connection request **/
+ private final String targetDataset;
+
+ private final FeedId receivingFeedId;
+
+
+ public FeedConnectionRequest(FeedJointKey feedPointKey, ConnectionLocation connectionLocation,
+ List<String> functionsToApply, String targetDataset, String policy, Map<String, String> policyParameters,
+ FeedId receivingFeedId) {
+ this.feedJointKey = feedPointKey;
+ this.connectionLocation = connectionLocation;
+ this.functionsToApply = functionsToApply;
+ this.targetDataset = targetDataset;
+ this.policy = policy;
+ this.policyParameters = policyParameters;
+ this.receivingFeedId = receivingFeedId;
+ this.connectionStatus = ConnectionStatus.INITIALIZED;
+ }
+
+ public FeedJointKey getFeedJointKey() {
+ return feedJointKey;
+ }
+
+ public ConnectionStatus getConnectionStatus() {
+ return connectionStatus;
+ }
+
+ public void setSubscriptionStatus(ConnectionStatus connectionStatus) {
+ this.connectionStatus = connectionStatus;
+ }
+
+ public String getPolicy() {
+ return policy;
+ }
+
+ public String getTargetDataset() {
+ return targetDataset;
+ }
+
+ public ConnectionLocation getSubscriptionLocation() {
+ return connectionLocation;
+ }
+
+ public FeedId getReceivingFeedId() {
+ return receivingFeedId;
+ }
+
+ public Map<String, String> getPolicyParameters() {
+ return policyParameters;
+ }
+
+ public List<String> getFunctionsToApply() {
+ return functionsToApply;
+ }
+
+ @Override
+ public String toString() {
+ return "Feed Connection Request " + feedJointKey + " [" + connectionLocation + "]" + " Apply ("
+ + StringUtils.join(functionsToApply, ",") + ")";
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/e800e6d5/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedId.java
----------------------------------------------------------------------
diff --git a/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedId.java b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedId.java
new file mode 100644
index 0000000..3145d72
--- /dev/null
+++ b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedId.java
@@ -0,0 +1,66 @@
+/*
+ * 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.management;
+
+import java.io.Serializable;
+
+/**
+ * A unique identifier for a data feed.
+ */
+public class FeedId implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String dataverse;
+ private final String feedName;
+
+ public FeedId(String dataverse, String feedName) {
+ this.dataverse = dataverse;
+ this.feedName = feedName;
+ }
+
+ public String getDataverse() {
+ return dataverse;
+ }
+
+ public String getFeedName() {
+ return feedName;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (o == null || !(o instanceof FeedId)) {
+ return false;
+ }
+ if (this == o || ((FeedId) o).getFeedName().equals(feedName) && ((FeedId) o).getDataverse().equals(dataverse)) {
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ public int hashCode() {
+ return toString().hashCode();
+ }
+
+ @Override
+ public String toString() {
+ return dataverse + "." + feedName;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/e800e6d5/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedInfo.java
----------------------------------------------------------------------
diff --git a/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedInfo.java b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedInfo.java
new file mode 100644
index 0000000..93f81d9
--- /dev/null
+++ b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedInfo.java
@@ -0,0 +1,53 @@
+/*
+ * 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.management;
+
+import org.apache.hyracks.api.job.JobId;
+import org.apache.hyracks.api.job.JobInfo;
+import org.apache.hyracks.api.job.JobSpecification;
+
+public class FeedInfo {
+ public JobSpecification jobSpec;
+ public JobInfo jobInfo;
+ public JobId jobId;
+ public FeedInfoType infoType;
+ public State state;
+
+ public enum State {
+ ACTIVE,
+ INACTIVE
+ }
+
+ public enum FeedInfoType {
+ INTAKE,
+ COLLECT
+ }
+
+ public FeedInfo(JobSpecification jobSpec, JobId jobId, FeedInfoType infoType) {
+ this.jobSpec = jobSpec;
+ this.jobId = jobId;
+ this.infoType = infoType;
+ this.state = State.INACTIVE;
+ }
+
+ @Override
+ public String toString() {
+ return " job id " + jobId;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/e800e6d5/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedJointKey.java
----------------------------------------------------------------------
diff --git a/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedJointKey.java b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedJointKey.java
new file mode 100644
index 0000000..7b08599
--- /dev/null
+++ b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedJointKey.java
@@ -0,0 +1,79 @@
+/*
+ * 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.management;
+
+import java.util.List;
+
+import org.apache.commons.lang3.StringUtils;
+
+/**
+ * Represents a unique identifier for a Feed Joint. A Feed joint is a logical entity located
+ * along a feed ingestion pipeline at a point where the tuples moving as part of data flow
+ * constitute the feed. The feed joint acts as a network tap and allows the flowing data to be
+ * routed to multiple paths.
+ */
+public class FeedJointKey {
+
+ private final FeedId primaryFeedId;
+ private final List<String> appliedFunctions;
+ private final String stringRep;
+
+ public FeedJointKey(FeedId feedId, List<String> appliedFunctions) {
+ this.primaryFeedId = feedId;
+ this.appliedFunctions = appliedFunctions;
+ StringBuilder builder = new StringBuilder();
+ builder.append(feedId);
+ builder.append(":");
+ builder.append(StringUtils.join(appliedFunctions, ':'));
+ stringRep = builder.toString();
+ }
+
+ public FeedId getFeedId() {
+ return primaryFeedId;
+ }
+
+ public List<String> getAppliedFunctions() {
+ return appliedFunctions;
+ }
+
+ public String getStringRep() {
+ return stringRep;
+ }
+
+ @Override
+ public final String toString() {
+ return stringRep;
+ }
+
+ @Override
+ public int hashCode() {
+ return stringRep.hashCode();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o)
+ return true;
+ if (o == null || !(o instanceof FeedJointKey)) {
+ return false;
+ }
+ return stringRep.equals(((FeedJointKey) o).stringRep);
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/e800e6d5/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedLifecycleEventSubscriber.java
----------------------------------------------------------------------
diff --git a/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedLifecycleEventSubscriber.java b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedLifecycleEventSubscriber.java
new file mode 100644
index 0000000..6e3cebc
--- /dev/null
+++ b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedLifecycleEventSubscriber.java
@@ -0,0 +1,66 @@
+/*
+ * 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.management;
+
+import java.util.Iterator;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import org.apache.asterix.common.exceptions.AsterixException;
+import org.apache.asterix.external.feed.api.IFeedLifecycleEventSubscriber;
+
+public class FeedLifecycleEventSubscriber implements IFeedLifecycleEventSubscriber {
+
+ private LinkedBlockingQueue<FeedLifecycleEvent> inbox;
+
+ public FeedLifecycleEventSubscriber() {
+ this.inbox = new LinkedBlockingQueue<FeedLifecycleEvent>();
+ }
+
+ @Override
+ public void handleFeedEvent(FeedLifecycleEvent event) {
+ inbox.add(event);
+ }
+
+ @Override
+ public void assertEvent(FeedLifecycleEvent event) throws AsterixException, InterruptedException {
+ boolean eventOccurred = false;
+ FeedLifecycleEvent e = null;
+ Iterator<FeedLifecycleEvent> eventsSoFar = inbox.iterator();
+ while (eventsSoFar.hasNext()) {
+ e = eventsSoFar.next();
+ assertNoFailure(e);
+ eventOccurred = e.equals(event);
+ }
+
+ while (!eventOccurred) {
+ e = inbox.take();
+ eventOccurred = e.equals(event);
+ if (!eventOccurred) {
+ assertNoFailure(e);
+ }
+ }
+ }
+
+ private void assertNoFailure(FeedLifecycleEvent e) throws AsterixException {
+ if (e.equals(FeedLifecycleEvent.FEED_INTAKE_FAILURE) || e.equals(FeedLifecycleEvent.FEED_COLLECT_FAILURE)) {
+ throw new AsterixException("Failure in feed");
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/e800e6d5/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedManager.java
----------------------------------------------------------------------
diff --git a/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedManager.java b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedManager.java
new file mode 100644
index 0000000..5095e7d
--- /dev/null
+++ b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedManager.java
@@ -0,0 +1,137 @@
+/*
+ * 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.management;
+
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import org.apache.asterix.common.config.AsterixFeedProperties;
+import org.apache.asterix.common.exceptions.AsterixException;
+import org.apache.asterix.external.feed.api.IFeedConnectionManager;
+import org.apache.asterix.external.feed.api.IFeedManager;
+import org.apache.asterix.external.feed.api.IFeedMemoryManager;
+import org.apache.asterix.external.feed.api.IFeedMessageService;
+import org.apache.asterix.external.feed.api.IFeedMetadataManager;
+import org.apache.asterix.external.feed.api.IFeedMetricCollector;
+import org.apache.asterix.external.feed.api.IFeedSubscriptionManager;
+import org.apache.asterix.external.feed.message.FeedMessageService;
+import org.apache.asterix.external.feed.watch.FeedMetricCollector;
+import org.apache.asterix.external.feed.watch.NodeLoadReportService;
+import org.apache.asterix.om.util.AsterixClusterProperties;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+
+/**
+ * An implementation of the IFeedManager interface.
+ * Provider necessary central repository for registering/retrieving
+ * artifacts/services associated with a feed.
+ */
+public class FeedManager implements IFeedManager {
+
+ private static final Logger LOGGER = Logger.getLogger(FeedManager.class.getName());
+
+ private final IFeedSubscriptionManager feedSubscriptionManager;
+
+ private final IFeedConnectionManager feedConnectionManager;
+
+ private final IFeedMemoryManager feedMemoryManager;
+
+ private final IFeedMetricCollector feedMetricCollector;
+
+ private final IFeedMetadataManager feedMetadataManager;
+
+ private final IFeedMessageService feedMessageService;
+
+ private final NodeLoadReportService nodeLoadReportService;
+
+ private final AsterixFeedProperties asterixFeedProperties;
+
+ private final String nodeId;
+
+ private final int frameSize;
+
+ public FeedManager(String nodeId, AsterixFeedProperties feedProperties, int frameSize)
+ throws AsterixException, HyracksDataException {
+ this.nodeId = nodeId;
+ this.feedSubscriptionManager = new FeedSubscriptionManager(nodeId);
+ this.feedConnectionManager = new FeedConnectionManager(nodeId);
+ this.feedMetadataManager = new FeedMetadataManager(nodeId);
+ this.feedMemoryManager = new FeedMemoryManager(nodeId, feedProperties, frameSize);
+ String ccClusterIp = AsterixClusterProperties.INSTANCE.getCluster() != null
+ ? AsterixClusterProperties.INSTANCE.getCluster().getMasterNode().getClusterIp() : "localhost";
+ this.feedMessageService = new FeedMessageService(feedProperties, nodeId, ccClusterIp);
+ this.nodeLoadReportService = new NodeLoadReportService(nodeId, this);
+ try {
+ this.feedMessageService.start();
+ this.nodeLoadReportService.start();
+ } catch (Exception e) {
+ if (LOGGER.isLoggable(Level.WARNING)) {
+ LOGGER.warning("Unable to start feed services " + e.getMessage());
+ }
+ e.printStackTrace();
+ }
+ this.feedMetricCollector = new FeedMetricCollector(nodeId);
+ this.frameSize = frameSize;
+ this.asterixFeedProperties = feedProperties;
+ }
+
+ @Override
+ public IFeedSubscriptionManager getFeedSubscriptionManager() {
+ return feedSubscriptionManager;
+ }
+
+ @Override
+ public IFeedConnectionManager getFeedConnectionManager() {
+ return feedConnectionManager;
+ }
+
+ @Override
+ public IFeedMemoryManager getFeedMemoryManager() {
+ return feedMemoryManager;
+ }
+
+ @Override
+ public IFeedMetricCollector getFeedMetricCollector() {
+ return feedMetricCollector;
+ }
+
+ public int getFrameSize() {
+ return frameSize;
+ }
+
+ @Override
+ public IFeedMetadataManager getFeedMetadataManager() {
+ return feedMetadataManager;
+ }
+
+ @Override
+ public IFeedMessageService getFeedMessageService() {
+ return feedMessageService;
+ }
+
+ @Override
+ public String toString() {
+ return "FeedManager " + "[" + nodeId + "]";
+ }
+
+ @Override
+ public AsterixFeedProperties getAsterixFeedProperties() {
+ return asterixFeedProperties;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/e800e6d5/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedMemoryManager.java
----------------------------------------------------------------------
diff --git a/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedMemoryManager.java b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedMemoryManager.java
new file mode 100644
index 0000000..de9d22c
--- /dev/null
+++ b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedMemoryManager.java
@@ -0,0 +1,114 @@
+/*
+ * 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.management;
+
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import org.apache.asterix.common.config.AsterixFeedProperties;
+import org.apache.asterix.external.feed.api.IFeedMemoryComponent;
+import org.apache.asterix.external.feed.api.IFeedMemoryManager;
+import org.apache.asterix.external.feed.api.IFeedMemoryComponent.Type;
+import org.apache.asterix.external.feed.dataflow.DataBucketPool;
+import org.apache.asterix.external.feed.dataflow.FrameCollection;
+
+public class FeedMemoryManager implements IFeedMemoryManager {
+
+ private static final Logger LOGGER = Logger.getLogger(FeedMemoryManager.class.getName());
+ private static final int ALLOCATION_INCREMENT = 10;
+
+ private final AtomicInteger componentId = new AtomicInteger(0);
+ private final String nodeId;
+ private final int budget;
+ private final int frameSize;
+
+ private int committed;
+
+ public FeedMemoryManager(String nodeId, AsterixFeedProperties feedProperties, int frameSize) {
+ this.nodeId = nodeId;
+ this.frameSize = frameSize;
+ budget = (int) feedProperties.getMemoryComponentGlobalBudget() / frameSize;
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("Feed Memory budget " + budget + " frames (frame size=" + frameSize + ")");
+ }
+ }
+
+ @Override
+ public synchronized IFeedMemoryComponent getMemoryComponent(Type type) {
+ IFeedMemoryComponent memoryComponent = null;
+ boolean valid = false;
+ switch (type) {
+ case COLLECTION:
+ valid = committed + START_COLLECTION_SIZE <= budget;
+ if (valid) {
+ memoryComponent = new FrameCollection(componentId.incrementAndGet(), this, START_COLLECTION_SIZE);
+ }
+ break;
+ case POOL:
+ valid = committed + START_POOL_SIZE <= budget;
+ if (valid) {
+ memoryComponent = new DataBucketPool(componentId.incrementAndGet(), this, START_POOL_SIZE,
+ frameSize);
+ }
+ committed += START_POOL_SIZE;
+ break;
+ }
+ if (!valid) {
+ if (LOGGER.isLoggable(Level.WARNING)) {
+ LOGGER.warning("Unable to allocate memory component of type" + type);
+ }
+ }
+ return valid ? memoryComponent : null;
+ }
+
+ @Override
+ public synchronized boolean expandMemoryComponent(IFeedMemoryComponent memoryComponent) {
+ if (committed + ALLOCATION_INCREMENT > budget) {
+ if (LOGGER.isLoggable(Level.WARNING)) {
+ LOGGER.warning("Memory budget " + budget + " is exhausted. Space left: " + (budget - committed)
+ + " frames.");
+ }
+ return false;
+ } else {
+ memoryComponent.expand(ALLOCATION_INCREMENT);
+ committed += ALLOCATION_INCREMENT;
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("Expanded memory component " + memoryComponent + " by " + ALLOCATION_INCREMENT + " " + this);
+ }
+ return true;
+ }
+ }
+
+ @Override
+ public synchronized void releaseMemoryComponent(IFeedMemoryComponent memoryComponent) {
+ int delta = memoryComponent.getTotalAllocation();
+ committed -= delta;
+ memoryComponent.reset();
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("Reset " + memoryComponent + " and reclaimed " + delta + " frames " + this);
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "FeedMemoryManager [" + nodeId + "]" + "(" + committed + "/" + budget + ")";
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/e800e6d5/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedMetadataManager.java
----------------------------------------------------------------------
diff --git a/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedMetadataManager.java b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedMetadataManager.java
new file mode 100644
index 0000000..34ae461
--- /dev/null
+++ b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedMetadataManager.java
@@ -0,0 +1,112 @@
+/*
+ * 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.management;
+
+import java.util.Date;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import org.apache.asterix.common.exceptions.AsterixException;
+import org.apache.asterix.external.feed.api.IFeedManager;
+import org.apache.asterix.external.feed.api.IFeedMetadataManager;
+import org.apache.asterix.external.feed.message.XAQLFeedMessage;
+import org.apache.asterix.external.util.FeedConstants;
+import org.apache.asterix.om.base.ARecord;
+import org.apache.asterix.om.base.AString;
+import org.apache.asterix.om.base.IAObject;
+import org.apache.asterix.om.types.ARecordType;
+import org.apache.asterix.om.types.BuiltinType;
+import org.apache.asterix.om.types.IAType;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+
+public class FeedMetadataManager implements IFeedMetadataManager {
+
+ private static final Logger LOGGER = Logger.getLogger(FeedMetadataManager.class.getName());
+
+ private final String nodeId;
+ private ARecordType recordType;
+
+ public FeedMetadataManager(String nodeId) throws AsterixException, HyracksDataException {
+ this.nodeId = nodeId;
+ String[] fieldNames = new String[] { "id", "dataverseName", "feedName", "targetDataset", "tuple", "message",
+ "timestamp" };
+ IAType[] fieldTypes = new IAType[] { BuiltinType.ASTRING, BuiltinType.ASTRING, BuiltinType.ASTRING,
+ BuiltinType.ASTRING, BuiltinType.ASTRING, BuiltinType.ASTRING, BuiltinType.ASTRING };
+
+ recordType = new ARecordType(FeedConstants.FAILED_TUPLE_DATASET_TYPE, fieldNames, fieldTypes, true);
+ }
+
+ @Override
+ public void logTuple(FeedConnectionId connectionId, String tuple, String message, IFeedManager feedManager)
+ throws AsterixException {
+ try {
+ AString id = new AString("1");
+ AString dataverseValue = new AString(connectionId.getFeedId().getDataverse());
+ AString feedValue = new AString(connectionId.getFeedId().getFeedName());
+ AString targetDatasetValue = new AString(connectionId.getDatasetName());
+ AString tupleValue = new AString(tuple);
+ AString messageValue = new AString(message);
+ AString dateTime = new AString(new Date().toString());
+
+ IAObject[] fields = new IAObject[] { id, dataverseValue, feedValue, targetDatasetValue, tupleValue,
+ messageValue, dateTime };
+ ARecord record = new ARecord(recordType, fields);
+ StringBuilder builder = new StringBuilder();
+ builder.append("use dataverse " + FeedConstants.FEEDS_METADATA_DV + ";" + "\n");
+ builder.append("insert into dataset " + FeedConstants.FAILED_TUPLE_DATASET + " ");
+ builder.append(" (" + recordToString(record) + ")");
+ builder.append(";");
+
+ XAQLFeedMessage xAqlMessage = new XAQLFeedMessage(connectionId, builder.toString());
+ feedManager.getFeedMessageService().sendMessage(xAqlMessage);
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info(" Sent " + xAqlMessage.toJSON());
+ }
+ } catch (Exception pe) {
+ throw new AsterixException(pe);
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "FeedMetadataManager [" + nodeId + "]";
+ }
+
+ private String recordToString(ARecord record) {
+ String[] fieldNames = record.getType().getFieldNames();
+ StringBuilder sb = new StringBuilder();
+ sb.append("{ ");
+ for (int i = 0; i < fieldNames.length; i++) {
+ if (i > 0) {
+ sb.append(", ");
+ }
+ sb.append("\"" + fieldNames[i] + "\"");
+ sb.append(": ");
+ switch (record.getType().getFieldTypes()[i].getTypeTag()) {
+ case STRING:
+ sb.append("\"" + ((AString) record.getValueByPos(i)).getStringValue() + "\"");
+ break;
+ default:
+ break;
+ }
+ }
+ sb.append(" }");
+ return sb.toString();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/e800e6d5/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedRuntimeManager.java
----------------------------------------------------------------------
diff --git a/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedRuntimeManager.java b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedRuntimeManager.java
new file mode 100644
index 0000000..f7e98f7
--- /dev/null
+++ b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedRuntimeManager.java
@@ -0,0 +1,83 @@
+/*
+ * 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.management;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import org.apache.asterix.external.feed.api.IFeedConnectionManager;
+import org.apache.asterix.external.feed.runtime.FeedRuntime;
+import org.apache.asterix.external.feed.runtime.FeedRuntimeId;
+
+public class FeedRuntimeManager {
+
+ private static Logger LOGGER = Logger.getLogger(FeedRuntimeManager.class.getName());
+
+ private final FeedConnectionId connectionId;
+ private final IFeedConnectionManager connectionManager;
+ private final Map<FeedRuntimeId, FeedRuntime> feedRuntimes;
+
+ private final ExecutorService executorService;
+
+ public FeedRuntimeManager(FeedConnectionId connectionId, IFeedConnectionManager feedConnectionManager) {
+ this.connectionId = connectionId;
+ this.feedRuntimes = new ConcurrentHashMap<FeedRuntimeId, FeedRuntime>();
+ this.executorService = Executors.newCachedThreadPool();
+ this.connectionManager = feedConnectionManager;
+ }
+
+ public void close() throws IOException {
+ if (executorService != null) {
+ executorService.shutdownNow();
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("Shut down executor service for :" + connectionId);
+ }
+ }
+ }
+
+ public FeedRuntime getFeedRuntime(FeedRuntimeId runtimeId) {
+ return feedRuntimes.get(runtimeId);
+ }
+
+ public void registerFeedRuntime(FeedRuntimeId runtimeId, FeedRuntime feedRuntime) {
+ feedRuntimes.put(runtimeId, feedRuntime);
+ }
+
+ public synchronized void deregisterFeedRuntime(FeedRuntimeId runtimeId) {
+ feedRuntimes.remove(runtimeId);
+ if (feedRuntimes.isEmpty()) {
+ connectionManager.deregisterFeed(connectionId);
+ }
+ }
+
+ public ExecutorService getExecutorService() {
+ return executorService;
+ }
+
+ public Set<FeedRuntimeId> getFeedRuntimes() {
+ return feedRuntimes.keySet();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/e800e6d5/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedSubscriptionManager.java
----------------------------------------------------------------------
diff --git a/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedSubscriptionManager.java b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedSubscriptionManager.java
new file mode 100644
index 0000000..e402f92
--- /dev/null
+++ b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedSubscriptionManager.java
@@ -0,0 +1,76 @@
+/*
+ * 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.management;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import org.apache.asterix.external.feed.api.IFeedSubscriptionManager;
+import org.apache.asterix.external.feed.api.ISubscribableRuntime;
+import org.apache.asterix.external.feed.runtime.SubscribableFeedRuntimeId;
+
+public class FeedSubscriptionManager implements IFeedSubscriptionManager {
+
+ private static Logger LOGGER = Logger.getLogger(FeedSubscriptionManager.class.getName());
+
+ private final String nodeId;
+
+ private final Map<SubscribableFeedRuntimeId, ISubscribableRuntime> subscribableRuntimes;
+
+ public FeedSubscriptionManager(String nodeId) {
+ this.nodeId = nodeId;
+ this.subscribableRuntimes = new HashMap<SubscribableFeedRuntimeId, ISubscribableRuntime>();
+ }
+
+ @Override
+ public void registerFeedSubscribableRuntime(ISubscribableRuntime subscribableRuntime) {
+ SubscribableFeedRuntimeId sid = (SubscribableFeedRuntimeId) subscribableRuntime.getRuntimeId();
+ if (!subscribableRuntimes.containsKey(sid)) {
+ subscribableRuntimes.put(sid, subscribableRuntime);
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("Registered feed subscribable runtime " + subscribableRuntime);
+ }
+ } else {
+ if (LOGGER.isLoggable(Level.WARNING)) {
+ LOGGER.warning("Feed ingestion runtime " + subscribableRuntime + " already registered.");
+ }
+ }
+ }
+
+ @Override
+ public ISubscribableRuntime getSubscribableRuntime(SubscribableFeedRuntimeId subscribableFeedRuntimeId) {
+ return subscribableRuntimes.get(subscribableFeedRuntimeId);
+ }
+
+ @Override
+ public void deregisterFeedSubscribableRuntime(SubscribableFeedRuntimeId ingestionId) {
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("De-registered feed subscribable runtime " + ingestionId);
+ }
+ subscribableRuntimes.remove(ingestionId);
+ }
+
+ @Override
+ public String toString() {
+ return "IngestionManager [" + nodeId + "]";
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/e800e6d5/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedWorkManager.java
----------------------------------------------------------------------
diff --git a/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedWorkManager.java b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedWorkManager.java
new file mode 100644
index 0000000..f2491db
--- /dev/null
+++ b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/management/FeedWorkManager.java
@@ -0,0 +1,50 @@
+/*
+ * 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.management;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import org.apache.asterix.external.feed.api.IFeedWork;
+import org.apache.asterix.external.feed.api.IFeedWorkEventListener;
+import org.apache.asterix.external.feed.api.IFeedWorkManager;
+
+/**
+ * Handles asynchronous execution of feed management related tasks.
+ */
+public class FeedWorkManager implements IFeedWorkManager {
+
+ public static final FeedWorkManager INSTANCE = new FeedWorkManager();
+
+ private final ExecutorService executorService = Executors.newCachedThreadPool();
+
+ private FeedWorkManager() {
+ }
+
+ public void submitWork(IFeedWork work, IFeedWorkEventListener listener) {
+ Runnable runnable = work.getRunnable();
+ try {
+ executorService.execute(runnable);
+ listener.workCompleted(work);
+ } catch (Exception e) {
+ listener.workFailed(work, e);
+ }
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/e800e6d5/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/EndFeedMessage.java
----------------------------------------------------------------------
diff --git a/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/EndFeedMessage.java b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/EndFeedMessage.java
new file mode 100644
index 0000000..b0f7624
--- /dev/null
+++ b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/EndFeedMessage.java
@@ -0,0 +1,96 @@
+/*
+ * 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.message;
+
+import org.json.JSONException;
+import org.json.JSONObject;
+import org.apache.asterix.external.feed.api.IFeedRuntime.FeedRuntimeType;
+import org.apache.asterix.external.feed.management.FeedConnectionId;
+import org.apache.asterix.external.feed.management.FeedId;
+import org.apache.asterix.external.util.FeedConstants;
+
+/**
+ * A feed control message indicating the need to end the feed. This message is dispatched
+ * to all locations that host an operator involved in the feed pipeline.
+ */
+public class EndFeedMessage extends FeedMessage {
+
+ private static final long serialVersionUID = 1L;
+
+ private final FeedId sourceFeedId;
+
+ private final FeedConnectionId connectionId;
+
+ private final FeedRuntimeType sourceRuntimeType;
+
+ private final boolean completeDisconnection;
+
+ private final EndMessageType endMessageType;
+
+ public enum EndMessageType {
+ DISCONNECT_FEED,
+ DISCONTINUE_SOURCE
+ }
+
+ public EndFeedMessage(FeedConnectionId connectionId, FeedRuntimeType sourceRuntimeType, FeedId sourceFeedId,
+ boolean completeDisconnection, EndMessageType endMessageType) {
+ super(MessageType.END);
+ this.connectionId = connectionId;
+ this.sourceRuntimeType = sourceRuntimeType;
+ this.sourceFeedId = sourceFeedId;
+ this.completeDisconnection = completeDisconnection;
+ this.endMessageType = endMessageType;
+ }
+
+ @Override
+ public String toString() {
+ return MessageType.END.name() + " " + connectionId + " [" + sourceRuntimeType + "] ";
+ }
+
+ public FeedRuntimeType getSourceRuntimeType() {
+ return sourceRuntimeType;
+ }
+
+ public FeedId getSourceFeedId() {
+ return sourceFeedId;
+ }
+
+ public boolean isCompleteDisconnection() {
+ return completeDisconnection;
+ }
+
+ public EndMessageType getEndMessageType() {
+ return endMessageType;
+ }
+
+ @Override
+ public JSONObject toJSON() throws JSONException {
+ JSONObject obj = new JSONObject();
+ obj.put(FeedConstants.MessageConstants.MESSAGE_TYPE, messageType.name());
+ obj.put(FeedConstants.MessageConstants.DATAVERSE, connectionId.getFeedId().getDataverse());
+ obj.put(FeedConstants.MessageConstants.FEED, connectionId.getFeedId().getFeedName());
+ obj.put(FeedConstants.MessageConstants.DATASET, connectionId.getDatasetName());
+ return obj;
+ }
+
+ public FeedConnectionId getFeedConnectionId() {
+ return connectionId;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/e800e6d5/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/FeedCongestionMessage.java
----------------------------------------------------------------------
diff --git a/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/FeedCongestionMessage.java b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/FeedCongestionMessage.java
new file mode 100644
index 0000000..6c924d2
--- /dev/null
+++ b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/FeedCongestionMessage.java
@@ -0,0 +1,102 @@
+/*
+ * 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.message;
+
+import org.json.JSONException;
+import org.json.JSONObject;
+import org.apache.asterix.external.feed.api.IFeedRuntime.FeedRuntimeType;
+import org.apache.asterix.external.feed.api.IFeedRuntime.Mode;
+import org.apache.asterix.external.feed.management.FeedConnectionId;
+import org.apache.asterix.external.feed.management.FeedId;
+import org.apache.asterix.external.feed.runtime.FeedRuntimeId;
+import org.apache.asterix.external.util.FeedConstants;
+import org.apache.asterix.external.util.FeedConstants.MessageConstants;
+
+public class FeedCongestionMessage extends FeedMessage {
+
+ private static final long serialVersionUID = 1L;
+
+ private final FeedConnectionId connectionId;
+ private final FeedRuntimeId runtimeId;
+ private int inflowRate;
+ private int outflowRate;
+ private Mode mode;
+
+ public FeedCongestionMessage(FeedConnectionId connectionId, FeedRuntimeId runtimeId, int inflowRate,
+ int outflowRate, Mode mode) {
+ super(MessageType.CONGESTION);
+ this.connectionId = connectionId;
+ this.runtimeId = runtimeId;
+ this.inflowRate = inflowRate;
+ this.outflowRate = outflowRate;
+ this.mode = mode;
+ }
+
+ @Override
+ public JSONObject toJSON() throws JSONException {
+ JSONObject obj = new JSONObject();
+ obj.put(FeedConstants.MessageConstants.MESSAGE_TYPE, messageType.name());
+ obj.put(FeedConstants.MessageConstants.DATAVERSE, connectionId.getFeedId().getDataverse());
+ obj.put(FeedConstants.MessageConstants.FEED, connectionId.getFeedId().getFeedName());
+ obj.put(FeedConstants.MessageConstants.DATASET, connectionId.getDatasetName());
+ obj.put(FeedConstants.MessageConstants.RUNTIME_TYPE, runtimeId.getFeedRuntimeType());
+ obj.put(FeedConstants.MessageConstants.OPERAND_ID, runtimeId.getOperandId());
+ obj.put(FeedConstants.MessageConstants.PARTITION, runtimeId.getPartition());
+ obj.put(FeedConstants.MessageConstants.INFLOW_RATE, inflowRate);
+ obj.put(FeedConstants.MessageConstants.OUTFLOW_RATE, outflowRate);
+ obj.put(FeedConstants.MessageConstants.MODE, mode);
+ return obj;
+ }
+
+ public FeedRuntimeId getRuntimeId() {
+ return runtimeId;
+ }
+
+ public int getInflowRate() {
+ return inflowRate;
+ }
+
+ public int getOutflowRate() {
+ return outflowRate;
+ }
+
+ public static FeedCongestionMessage read(JSONObject obj) throws JSONException {
+ FeedId feedId = new FeedId(obj.getString(FeedConstants.MessageConstants.DATAVERSE),
+ obj.getString(FeedConstants.MessageConstants.FEED));
+ FeedConnectionId connectionId = new FeedConnectionId(feedId,
+ obj.getString(FeedConstants.MessageConstants.DATASET));
+ FeedRuntimeId runtimeId = new FeedRuntimeId(FeedRuntimeType.valueOf(obj
+ .getString(FeedConstants.MessageConstants.RUNTIME_TYPE)),
+ obj.getInt(FeedConstants.MessageConstants.PARTITION),
+ obj.getString(FeedConstants.MessageConstants.OPERAND_ID));
+ Mode mode = Mode.valueOf(obj.getString(MessageConstants.MODE));
+ return new FeedCongestionMessage(connectionId, runtimeId,
+ obj.getInt(FeedConstants.MessageConstants.INFLOW_RATE),
+ obj.getInt(FeedConstants.MessageConstants.OUTFLOW_RATE), mode);
+ }
+
+ public FeedConnectionId getConnectionId() {
+ return connectionId;
+ }
+
+ public Mode getMode() {
+ return mode;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/e800e6d5/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/FeedMessage.java
----------------------------------------------------------------------
diff --git a/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/FeedMessage.java b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/FeedMessage.java
new file mode 100644
index 0000000..f2b354b
--- /dev/null
+++ b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/FeedMessage.java
@@ -0,0 +1,42 @@
+/*
+ * 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.message;
+
+import org.apache.asterix.external.feed.api.IFeedMessage;
+import org.apache.hyracks.api.dataflow.value.JSONSerializable;
+
+/**
+ * A control message that can be sent to the runtime instance of a
+ * feed's adapter.
+ */
+public abstract class FeedMessage implements IFeedMessage, JSONSerializable {
+
+ private static final long serialVersionUID = 1L;
+
+ protected final MessageType messageType;
+
+ public FeedMessage(MessageType messageType) {
+ this.messageType = messageType;
+ }
+
+ public MessageType getMessageType() {
+ return messageType;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/e800e6d5/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/FeedMessageService.java
----------------------------------------------------------------------
diff --git a/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/FeedMessageService.java b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/FeedMessageService.java
new file mode 100644
index 0000000..13d6622
--- /dev/null
+++ b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/FeedMessageService.java
@@ -0,0 +1,145 @@
+/*
+ * 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.message;
+
+import java.net.Socket;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import org.json.JSONException;
+import org.json.JSONObject;
+
+import org.apache.asterix.common.config.AsterixFeedProperties;
+import org.apache.asterix.external.feed.api.IFeedMessage;
+import org.apache.asterix.external.feed.api.IFeedMessageService;
+import org.apache.asterix.external.util.FeedConstants;
+
+/**
+ * Sends feed report messages on behalf of an operator instance
+ * to the SuperFeedManager associated with the feed.
+ */
+public class FeedMessageService implements IFeedMessageService {
+
+ private static final Logger LOGGER = Logger.getLogger(FeedMessageService.class.getName());
+
+ private final LinkedBlockingQueue<String> inbox;
+ private final FeedMessageHandler mesgHandler;
+ private final String nodeId;
+ private ExecutorService executor;
+
+ public FeedMessageService(AsterixFeedProperties feedProperties, String nodeId, String ccClusterIp) {
+ this.inbox = new LinkedBlockingQueue<String>();
+ this.mesgHandler = new FeedMessageHandler(inbox, ccClusterIp, feedProperties.getFeedCentralManagerPort());
+ this.nodeId = nodeId;
+ this.executor = Executors.newSingleThreadExecutor();
+ }
+
+ public void start() throws Exception {
+
+ executor.execute(mesgHandler);
+ }
+
+ public void stop() {
+ synchronized (mesgHandler.getLock()) {
+ executor.shutdownNow();
+ }
+ mesgHandler.stop();
+ }
+
+ @Override
+ public void sendMessage(IFeedMessage message) {
+ try {
+ JSONObject obj = message.toJSON();
+ obj.put(FeedConstants.MessageConstants.NODE_ID, nodeId);
+ obj.put(FeedConstants.MessageConstants.MESSAGE_TYPE, message.getMessageType().name());
+ inbox.add(obj.toString());
+ } catch (JSONException jse) {
+ if (LOGGER.isLoggable(Level.WARNING)) {
+ LOGGER.warning("JSON exception in parsing message " + message + " exception [" + jse.getMessage() + "]");
+ }
+ }
+ }
+
+ private static class FeedMessageHandler implements Runnable {
+
+ private final LinkedBlockingQueue<String> inbox;
+ private final String host;
+ private final int port;
+ private final Object lock;
+
+ private Socket cfmSocket;
+
+ private static final byte[] EOL = "\n".getBytes();
+
+ public FeedMessageHandler(LinkedBlockingQueue<String> inbox, String host, int port) {
+ this.inbox = inbox;
+ this.host = host;
+ this.port = port;
+ this.lock = new Object();
+ }
+
+ public void run() {
+ try {
+ cfmSocket = new Socket(host, port);
+ if (cfmSocket != null) {
+ while (true) {
+ String message = inbox.take();
+ synchronized (lock) { // lock prevents message handler from sending incomplete message midst shutdown attempt
+ cfmSocket.getOutputStream().write(message.getBytes());
+ cfmSocket.getOutputStream().write(EOL);
+ }
+ }
+ } else {
+ if (LOGGER.isLoggable(Level.WARNING)) {
+ LOGGER.warning("Unable to start feed message service");
+ }
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ if (LOGGER.isLoggable(Level.WARNING)) {
+ LOGGER.warning("Exception in handling incoming feed messages" + e.getMessage());
+ }
+ } finally {
+ stop();
+ }
+
+ }
+
+ public void stop() {
+ if (cfmSocket != null) {
+ try {
+ cfmSocket.close();
+ } catch (Exception e) {
+ if (LOGGER.isLoggable(Level.WARNING)) {
+ LOGGER.warning("Exception in closing socket " + e.getMessage());
+ }
+ }
+ }
+ }
+
+ public Object getLock() {
+ return lock;
+ }
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/e800e6d5/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/FeedReportMessage.java
----------------------------------------------------------------------
diff --git a/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/FeedReportMessage.java b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/FeedReportMessage.java
new file mode 100644
index 0000000..1b8c45d
--- /dev/null
+++ b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/FeedReportMessage.java
@@ -0,0 +1,99 @@
+/*
+ * 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.message;
+
+import org.json.JSONException;
+import org.json.JSONObject;
+import org.apache.asterix.external.feed.api.IFeedMetricCollector.ValueType;
+import org.apache.asterix.external.feed.api.IFeedRuntime.FeedRuntimeType;
+import org.apache.asterix.external.feed.management.FeedConnectionId;
+import org.apache.asterix.external.feed.management.FeedId;
+import org.apache.asterix.external.feed.runtime.FeedRuntimeId;
+import org.apache.asterix.external.util.FeedConstants;
+import org.apache.asterix.external.util.FeedConstants.MessageConstants;
+
+public class FeedReportMessage extends FeedMessage {
+
+ private static final long serialVersionUID = 1L;
+
+ private final FeedConnectionId connectionId;
+ private final FeedRuntimeId runtimeId;
+ private final ValueType valueType;
+ private int value;
+
+ public FeedReportMessage(FeedConnectionId connectionId, FeedRuntimeId runtimeId, ValueType valueType, int value) {
+ super(MessageType.FEED_REPORT);
+ this.connectionId = connectionId;
+ this.runtimeId = runtimeId;
+ this.valueType = valueType;
+ this.value = value;
+ }
+
+ public void reset(int value) {
+ this.value = value;
+ }
+
+ @Override
+ public JSONObject toJSON() throws JSONException {
+ JSONObject obj = new JSONObject();
+ obj.put(FeedConstants.MessageConstants.MESSAGE_TYPE, messageType.name());
+ obj.put(FeedConstants.MessageConstants.DATAVERSE, connectionId.getFeedId().getDataverse());
+ obj.put(FeedConstants.MessageConstants.FEED, connectionId.getFeedId().getFeedName());
+ obj.put(FeedConstants.MessageConstants.DATASET, connectionId.getDatasetName());
+ obj.put(FeedConstants.MessageConstants.RUNTIME_TYPE, runtimeId.getFeedRuntimeType());
+ obj.put(FeedConstants.MessageConstants.PARTITION, runtimeId.getPartition());
+ obj.put(FeedConstants.MessageConstants.VALUE_TYPE, valueType);
+ obj.put(FeedConstants.MessageConstants.VALUE, value);
+ return obj;
+ }
+
+ public static FeedReportMessage read(JSONObject obj) throws JSONException {
+ FeedId feedId = new FeedId(obj.getString(FeedConstants.MessageConstants.DATAVERSE),
+ obj.getString(FeedConstants.MessageConstants.FEED));
+ FeedConnectionId connectionId = new FeedConnectionId(feedId,
+ obj.getString(FeedConstants.MessageConstants.DATASET));
+ FeedRuntimeId runtimeId = new FeedRuntimeId(FeedRuntimeType.valueOf(obj
+ .getString(FeedConstants.MessageConstants.RUNTIME_TYPE)),
+ obj.getInt(FeedConstants.MessageConstants.PARTITION), FeedConstants.MessageConstants.NOT_APPLICABLE);
+ ValueType type = ValueType.valueOf(obj.getString(MessageConstants.VALUE_TYPE));
+ int value = Integer.parseInt(obj.getString(MessageConstants.VALUE));
+ return new FeedReportMessage(connectionId, runtimeId, type, value);
+ }
+
+ public int getValue() {
+ return value;
+ }
+
+ public void setValue(int value) {
+ this.value = value;
+ }
+
+ public FeedConnectionId getConnectionId() {
+ return connectionId;
+ }
+
+ public FeedRuntimeId getRuntimeId() {
+ return runtimeId;
+ }
+
+ public ValueType getValueType() {
+ return valueType;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/e800e6d5/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/FeedTupleCommitAckMessage.java
----------------------------------------------------------------------
diff --git a/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/FeedTupleCommitAckMessage.java b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/FeedTupleCommitAckMessage.java
new file mode 100644
index 0000000..61e26de
--- /dev/null
+++ b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/FeedTupleCommitAckMessage.java
@@ -0,0 +1,98 @@
+/*
+ * 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.message;
+
+import javax.xml.bind.DatatypeConverter;
+
+import org.json.JSONException;
+import org.json.JSONObject;
+import org.apache.asterix.external.feed.management.FeedConnectionId;
+import org.apache.asterix.external.feed.management.FeedId;
+import org.apache.asterix.external.util.FeedConstants;
+
+public class FeedTupleCommitAckMessage extends FeedMessage {
+
+ private static final long serialVersionUID = 1L;
+
+ private final FeedConnectionId connectionId;
+ private int intakePartition;
+ private int base;
+ private byte[] commitAcks;
+
+ public FeedTupleCommitAckMessage(FeedConnectionId connectionId, int intakePartition, int base, byte[] commitAcks) {
+ super(MessageType.COMMIT_ACK);
+ this.connectionId = connectionId;
+ this.intakePartition = intakePartition;
+ this.base = base;
+ this.commitAcks = commitAcks;
+ }
+
+ @Override
+ public JSONObject toJSON() throws JSONException {
+ JSONObject obj = new JSONObject();
+ obj.put(FeedConstants.MessageConstants.MESSAGE_TYPE, messageType.name());
+ obj.put(FeedConstants.MessageConstants.DATAVERSE, connectionId.getFeedId().getDataverse());
+ obj.put(FeedConstants.MessageConstants.FEED, connectionId.getFeedId().getFeedName());
+ obj.put(FeedConstants.MessageConstants.DATASET, connectionId.getDatasetName());
+ obj.put(FeedConstants.MessageConstants.INTAKE_PARTITION, intakePartition);
+ obj.put(FeedConstants.MessageConstants.BASE, base);
+ String commitAcksString = DatatypeConverter.printBase64Binary(commitAcks);
+ obj.put(FeedConstants.MessageConstants.COMMIT_ACKS, commitAcksString);
+ return obj;
+ }
+
+ public static FeedTupleCommitAckMessage read(JSONObject obj) throws JSONException {
+ FeedId feedId = new FeedId(obj.getString(FeedConstants.MessageConstants.DATAVERSE),
+ obj.getString(FeedConstants.MessageConstants.FEED));
+ FeedConnectionId connectionId = new FeedConnectionId(feedId,
+ obj.getString(FeedConstants.MessageConstants.DATASET));
+ int intakePartition = obj.getInt(FeedConstants.MessageConstants.INTAKE_PARTITION);
+ int base = obj.getInt(FeedConstants.MessageConstants.BASE);
+ String commitAcksString = obj.getString(FeedConstants.MessageConstants.COMMIT_ACKS);
+ byte[] commitAcks = DatatypeConverter.parseBase64Binary(commitAcksString);
+ return new FeedTupleCommitAckMessage(connectionId, intakePartition, base, commitAcks);
+ }
+
+ public FeedConnectionId getConnectionId() {
+ return connectionId;
+ }
+
+ public int getIntakePartition() {
+ return intakePartition;
+ }
+
+ public byte[] getCommitAcks() {
+ return commitAcks;
+ }
+
+ public void reset(int intakePartition, int base, byte[] commitAcks) {
+ this.intakePartition = intakePartition;
+ this.base = base;
+ this.commitAcks = commitAcks;
+ }
+
+ public int getBase() {
+ return base;
+ }
+
+ public void setBase(int base) {
+ this.base = base;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/e800e6d5/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/FeedTupleCommitResponseMessage.java
----------------------------------------------------------------------
diff --git a/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/FeedTupleCommitResponseMessage.java b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/FeedTupleCommitResponseMessage.java
new file mode 100644
index 0000000..a61dc06
--- /dev/null
+++ b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/FeedTupleCommitResponseMessage.java
@@ -0,0 +1,81 @@
+/*
+ * 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.message;
+
+import org.json.JSONException;
+import org.json.JSONObject;
+import org.apache.asterix.external.feed.management.FeedConnectionId;
+import org.apache.asterix.external.feed.management.FeedId;
+import org.apache.asterix.external.util.FeedConstants;
+
+public class FeedTupleCommitResponseMessage extends FeedMessage {
+
+ private static final long serialVersionUID = 1L;
+
+ private final FeedConnectionId connectionId;
+ private final int intakePartition;
+ private final int maxWindowAcked;
+
+ public FeedTupleCommitResponseMessage(FeedConnectionId connectionId, int intakePartition, int maxWindowAcked) {
+ super(MessageType.COMMIT_ACK_RESPONSE);
+ this.connectionId = connectionId;
+ this.intakePartition = intakePartition;
+ this.maxWindowAcked = maxWindowAcked;
+ }
+
+ @Override
+ public JSONObject toJSON() throws JSONException {
+ JSONObject obj = new JSONObject();
+ obj.put(FeedConstants.MessageConstants.MESSAGE_TYPE, messageType.name());
+ obj.put(FeedConstants.MessageConstants.DATAVERSE, connectionId.getFeedId().getDataverse());
+ obj.put(FeedConstants.MessageConstants.FEED, connectionId.getFeedId().getFeedName());
+ obj.put(FeedConstants.MessageConstants.DATASET, connectionId.getDatasetName());
+ obj.put(FeedConstants.MessageConstants.INTAKE_PARTITION, intakePartition);
+ obj.put(FeedConstants.MessageConstants.MAX_WINDOW_ACKED, maxWindowAcked);
+ return obj;
+ }
+
+ @Override
+ public String toString() {
+ return connectionId + "[" + intakePartition + "]" + "(" + maxWindowAcked + ")";
+ }
+
+ public static FeedTupleCommitResponseMessage read(JSONObject obj) throws JSONException {
+ FeedId feedId = new FeedId(obj.getString(FeedConstants.MessageConstants.DATAVERSE),
+ obj.getString(FeedConstants.MessageConstants.FEED));
+ FeedConnectionId connectionId = new FeedConnectionId(feedId,
+ obj.getString(FeedConstants.MessageConstants.DATASET));
+ int intakePartition = obj.getInt(FeedConstants.MessageConstants.INTAKE_PARTITION);
+ int maxWindowAcked = obj.getInt(FeedConstants.MessageConstants.MAX_WINDOW_ACKED);
+ return new FeedTupleCommitResponseMessage(connectionId, intakePartition, maxWindowAcked);
+ }
+
+ public FeedConnectionId getConnectionId() {
+ return connectionId;
+ }
+
+ public int getMaxWindowAcked() {
+ return maxWindowAcked;
+ }
+
+ public int getIntakePartition() {
+ return intakePartition;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/e800e6d5/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/MessageListener.java
----------------------------------------------------------------------
diff --git a/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/MessageListener.java b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/MessageListener.java
new file mode 100644
index 0000000..67f2884
--- /dev/null
+++ b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/MessageListener.java
@@ -0,0 +1,126 @@
+/*
+ * 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.message;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.nio.CharBuffer;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+public class MessageListener {
+
+ private static final Logger LOGGER = Logger.getLogger(MessageListener.class.getName());
+
+ private int port;
+ private final LinkedBlockingQueue<String> outbox;
+
+ private ExecutorService executorService = Executors.newFixedThreadPool(10);
+
+ private MessageListenerServer listenerServer;
+
+ public MessageListener(int port, LinkedBlockingQueue<String> outbox) {
+ this.port = port;
+ this.outbox = outbox;
+ }
+
+ public void stop() {
+ listenerServer.stop();
+ if (!executorService.isShutdown()) {
+ executorService.shutdownNow();
+ }
+ }
+
+ public void start() throws IOException {
+ listenerServer = new MessageListenerServer(port, outbox);
+ executorService.execute(listenerServer);
+ }
+
+ private static class MessageListenerServer implements Runnable {
+
+ private final int port;
+ private final LinkedBlockingQueue<String> outbox;
+ private ServerSocket server;
+
+ private static final char EOL = (char) "\n".getBytes()[0];
+
+ public MessageListenerServer(int port, LinkedBlockingQueue<String> outbox) {
+ this.port = port;
+ this.outbox = outbox;
+ }
+
+ public void stop() {
+ try {
+ server.close();
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+
+ @Override
+ public void run() {
+ Socket client = null;
+ try {
+ server = new ServerSocket(port);
+ client = server.accept();
+ InputStream in = client.getInputStream();
+ CharBuffer buffer = CharBuffer.allocate(5000);
+ char ch;
+ while (true) {
+ ch = (char) in.read();
+ if (((int) ch) == -1) {
+ break;
+ }
+ while (ch != EOL) {
+ buffer.put(ch);
+ ch = (char) in.read();
+ }
+ buffer.flip();
+ String s = new String(buffer.array());
+ synchronized (outbox) {
+ outbox.add(s + "\n");
+ }
+ buffer.position(0);
+ buffer.limit(5000);
+ }
+
+ } catch (Exception e) {
+ if (LOGGER.isLoggable(Level.WARNING)) {
+ LOGGER.warning("Unable to start Message listener" + server);
+ }
+ } finally {
+ if (server != null) {
+ try {
+ server.close();
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+ }
+
+ }
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/e800e6d5/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/MessageReceiver.java
----------------------------------------------------------------------
diff --git a/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/MessageReceiver.java b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/MessageReceiver.java
new file mode 100644
index 0000000..abeb994
--- /dev/null
+++ b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/MessageReceiver.java
@@ -0,0 +1,111 @@
+/*
+ * 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.message;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import org.apache.asterix.external.feed.api.IMessageReceiver;
+
+public abstract class MessageReceiver<T> implements IMessageReceiver<T> {
+
+ protected static final Logger LOGGER = Logger.getLogger(MessageReceiver.class.getName());
+
+ protected final LinkedBlockingQueue<T> inbox;
+ protected ExecutorService executor;
+
+ public MessageReceiver() {
+ inbox = new LinkedBlockingQueue<T>();
+ }
+
+ public abstract void processMessage(T message) throws Exception;
+
+ @Override
+ public void start() {
+ executor = Executors.newSingleThreadExecutor();
+ executor.execute(new MessageReceiverRunnable<T>(this));
+ }
+
+ @Override
+ public synchronized void sendMessage(T message) {
+ inbox.add(message);
+ }
+
+ @Override
+ public void close(boolean processPending) {
+ if (executor != null) {
+ executor.shutdown();
+ executor = null;
+ if (processPending) {
+ flushPendingMessages();
+ } else {
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("Will discard the pending frames " + inbox.size());
+ }
+ }
+ }
+ }
+
+ private static class MessageReceiverRunnable<T> implements Runnable {
+
+ private final LinkedBlockingQueue<T> inbox;
+ private final MessageReceiver<T> messageReceiver;
+
+ public MessageReceiverRunnable(MessageReceiver<T> messageReceiver) {
+ this.inbox = messageReceiver.inbox;
+ this.messageReceiver = messageReceiver;
+ }
+
+ @Override
+ public void run() {
+ while (true) {
+ try {
+ T message = inbox.take();
+ messageReceiver.processMessage(message);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+ }
+ }
+
+ protected void flushPendingMessages() {
+ while (!inbox.isEmpty()) {
+ T message = null;
+ try {
+ message = inbox.take();
+ processMessage(message);
+ } catch (InterruptedException ie) {
+ // ignore exception but break from the loop
+ break;
+ } catch (Exception e) {
+ e.printStackTrace();
+ if (LOGGER.isLoggable(Level.WARNING)) {
+ LOGGER.warning("Exception " + e + " in processing message " + message);
+ }
+ }
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/e800e6d5/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/NodeReportMessage.java
----------------------------------------------------------------------
diff --git a/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/NodeReportMessage.java b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/NodeReportMessage.java
new file mode 100644
index 0000000..1548d6d
--- /dev/null
+++ b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/NodeReportMessage.java
@@ -0,0 +1,68 @@
+/*
+ * 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.message;
+
+import org.json.JSONException;
+import org.json.JSONObject;
+import org.apache.asterix.external.feed.api.IFeedMessage;
+import org.apache.asterix.external.util.FeedConstants;
+
+public class NodeReportMessage extends FeedMessage {
+
+ private static final long serialVersionUID = 1L;
+
+ private double cpuLoad;
+ private double usedHeap;
+ private int nRuntimes;
+
+ public NodeReportMessage(float cpuLoad, long usedHeap, int nRuntimes) {
+ super(IFeedMessage.MessageType.NODE_REPORT);
+ this.usedHeap = usedHeap;
+ this.cpuLoad = cpuLoad;
+ this.nRuntimes = nRuntimes;
+ }
+
+ public void reset(double cpuLoad, double usedHeap, int nRuntimes) {
+ this.cpuLoad = cpuLoad;
+ this.usedHeap = usedHeap;
+ this.nRuntimes = nRuntimes;
+ }
+
+ @Override
+ public JSONObject toJSON() throws JSONException {
+ JSONObject obj = new JSONObject();
+ obj.put(FeedConstants.MessageConstants.CPU_LOAD, cpuLoad);
+ obj.put(FeedConstants.MessageConstants.HEAP_USAGE, usedHeap);
+ obj.put(FeedConstants.MessageConstants.N_RUNTIMES, nRuntimes);
+ return obj;
+ }
+
+ public double getCpuLoad() {
+ return cpuLoad;
+ }
+
+ public double getUsedHeap() {
+ return usedHeap;
+ }
+
+ public int getnRuntimes() {
+ return nRuntimes;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-asterixdb/blob/e800e6d5/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/PrepareStallMessage.java
----------------------------------------------------------------------
diff --git a/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/PrepareStallMessage.java b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/PrepareStallMessage.java
new file mode 100644
index 0000000..76fe0c2
--- /dev/null
+++ b/asterix-external-data/src/main/java/org/apache/asterix/external/feed/message/PrepareStallMessage.java
@@ -0,0 +1,68 @@
+/*
+ * 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.message;
+
+import org.json.JSONException;
+import org.json.JSONObject;
+import org.apache.asterix.external.feed.management.FeedConnectionId;
+import org.apache.asterix.external.util.FeedConstants;
+
+/**
+ * A feed control message indicating the need to end the feed. This message is dispatched
+ * to all locations that host an operator involved in the feed pipeline.
+ */
+public class PrepareStallMessage extends FeedMessage {
+
+ private static final long serialVersionUID = 1L;
+
+ private final FeedConnectionId connectionId;
+
+ private final int computePartitionsRetainLimit;
+
+ public PrepareStallMessage(FeedConnectionId connectionId, int computePartitionsRetainLimit) {
+ super(MessageType.PREPARE_STALL);
+ this.connectionId = connectionId;
+ this.computePartitionsRetainLimit = computePartitionsRetainLimit;
+ }
+
+ @Override
+ public String toString() {
+ return MessageType.PREPARE_STALL.name() + " " + connectionId;
+ }
+
+ @Override
+ public JSONObject toJSON() throws JSONException {
+ JSONObject obj = new JSONObject();
+ obj.put(FeedConstants.MessageConstants.MESSAGE_TYPE, messageType.name());
+ obj.put(FeedConstants.MessageConstants.DATAVERSE, connectionId.getFeedId().getDataverse());
+ obj.put(FeedConstants.MessageConstants.FEED, connectionId.getFeedId().getFeedName());
+ obj.put(FeedConstants.MessageConstants.DATASET, connectionId.getDatasetName());
+ obj.put(FeedConstants.MessageConstants.COMPUTE_PARTITION_RETAIN_LIMIT, computePartitionsRetainLimit);
+ return obj;
+ }
+
+ public FeedConnectionId getConnectionId() {
+ return connectionId;
+ }
+
+ public int getComputePartitionsRetainLimit() {
+ return computePartitionsRetainLimit;
+ }
+
+}