Add code for: [3] Lost responsibility detected

Author: MoeweX

ImplementationSubscriptionManagement.md

@@ -14,13 +14,13 @@
 
 ### [3] Lost responsibility detected
 
-- [ ] Background task to detect lost responsibilities for each keygroup for which subscriptions exist
-- [ ] Detected: run UpdateKeygroupSubscriptionsTask
+- [x] Background task to detect lost responsibilities for each keygroup for which subscriptions exist
+- [x] Detected: run UpdateKeygroupSubscriptionsTask
 
 ### [4] Client updates/deletes keygroup via node
 
-- [ ] Add client keygroup update/delete functionality incl. naming service validation
-- [ ] Naming service approves: run UpdateKeygroupConfigTask
+- [x] Add client keygroup update/delete functionality incl. naming service validation
+- [x] Naming service approves: run UpdateKeygroupConfigTask
 
 ### [5] Not interpretable message
 

src/main/java/communication/SubscriptionRegistry.java

@@ -2,7 +2,9 @@
 
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Map;
+import java.util.Set;
 
 import org.apache.log4j.Logger;
 
@@ -87,6 +89,10 @@ public synchronized void unsubscribeFromKeygroup(KeygroupID keygroupID) {
 			s.stopReception();
 		}
 	}
+	
+	public synchronized Set<KeygroupID> getSubscribedKeygroups() {
+		return new HashSet<>(activeSubscriptions.keySet());
+	}
 
 	/**
 	 * Subscriptions for the given keygroupID exist, a value for the given key is present.

src/main/java/control/FBase.java

@@ -96,6 +96,7 @@ public void startup(boolean announce) throws InterruptedException, ExecutionExce
 		taskmanager.startBackgroundCheckKeygroupConfigurationsOnUpdatesTask(0);
 		taskmanager.startDetectMissingHeartbeatsTask(0, 0);
 		taskmanager.startBackgroundDetectMissingResponsibility(0);
+		taskmanager.startBackgroundDetectLostResponsibility(0);
 
 		Thread.sleep(50);
 		logger.info("FBase started, all background tasks up and running.");

src/main/java/tasks/TaskManager.java

@@ -16,6 +16,7 @@
 import model.data.DataRecord;
 import model.messages.Envelope;
 import tasks.background.CheckKeygroupConfigurationsOnUpdatesTask;
+import tasks.background.DetectLostResponsibility;
 import tasks.background.DetectMissingHeartbeats;
 import tasks.background.DetectMissingResponsibility;
 import tasks.background.PollLatestConfigurationDataForResponsibleKeygroupsTask;
@@ -41,7 +42,8 @@ public enum TaskName {
 		PROCESS_MESSAGE_WITH_UNKNOWN_ENCRYPTION, CHECK_NAMING_SERVICE_CONFIGURATION_DATA,
 		B_POLL_LATEST_CONFIGURATION_DATA_FOR_RESPONSIBLE_KEYGROUPS, B_PUT_HEARTBEAT,
 		B_DETECT_MISSING_HEARTBEATS, REMOVE_MACHINE_FROM_NODE,
-		ANNOUNCE_UPDATE_OF_OWN_NODE_CONFIGURATION, B_DETECT_MISSING_RESPONSIBILITY
+		ANNOUNCE_UPDATE_OF_OWN_NODE_CONFIGURATION, B_DETECT_MISSING_RESPONSIBILITY,
+		B_DETECT_LOST_RESPONSIBILITY
 	}
 
 	public void storeHistory() {
@@ -182,4 +184,9 @@ public Future<Boolean> startBackgroundDetectMissingResponsibility(int interval)
 		return future;
 	}
 
+	public Future<Boolean> startBackgroundDetectLostResponsibility(int interval) {
+		Future<Boolean> future = pool.submit(new DetectLostResponsibility(fBase, interval));
+		return future;
+	}
+
 }

src/main/java/tasks/background/DetectLostResponsibility.java

@@ -0,0 +1,96 @@
+package tasks.background;
+
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.log4j.Level;
+import org.apache.log4j.Logger;
+import org.javatuples.Pair;
+
+import control.FBase;
+import exceptions.FBaseException;
+import exceptions.FBaseStorageConnectorException;
+import model.config.KeygroupConfig;
+import model.data.KeygroupID;
+import tasks.Task;
+import tasks.TaskManager.TaskName;
+
+/**
+ * 
+ * This background task detects if a keygroup a machine created subscriptions for is not in
+ * the machine's responsibility anymore
+ * 
+ * @author jonathanhasenburg
+ *
+ */
+public class DetectLostResponsibility extends Task<Boolean> {
+
+	private static Logger logger = Logger.getLogger(DetectLostResponsibility.class.getName());
+
+	static {
+		logger.setLevel(Level.INFO);
+	}
+
+	/**
+	 * Creates a new {@link DetectLostResponsibility}. If checkInterval <= 0, the default is
+	 * used (10 sec).
+	 * 
+	 * @param fBase
+	 * @param checkInterval - the interval to put heartbeats in milliseconds
+	 */
+	public DetectLostResponsibility(FBase fBase, int checkInterval) {
+		super(TaskName.B_DETECT_LOST_RESPONSIBILITY, fBase);
+		if (checkInterval > 0) {
+			this.checkInterval = checkInterval;
+		}
+
+	}
+
+	private int checkInterval = 10000;
+
+	@Override
+	public Boolean executeFunctionality() {
+
+		while (!Thread.currentThread().isInterrupted()) {
+			logger.info("Looking for lost responsibilities");
+
+			try {
+				Set<KeygroupID> subscribedKeygroups =
+						fBase.subscriptionRegistry.getSubscribedKeygroups();
+
+				Map<KeygroupID, Pair<String, Integer>> responsibilities =
+						fBase.connector.keyGroupSubscriberMachines_listAll();
+
+				for (KeygroupID keygroupID : subscribedKeygroups) {
+					try {
+						Pair<String, Integer> pair = responsibilities.get(keygroupID);
+						if (pair == null
+								|| !pair.getValue0().equals(fBase.configuration.getMachineName())) {
+							logger.info("Lost responsibility for keygroup " + keygroupID);
+							KeygroupConfig config = fBase.connector.keygroupConfig_get(keygroupID);
+							fBase.taskmanager.runUpdateKeygroupSubscriptionsTask(config);
+						}
+					} catch (FBaseException e) {
+						logger.error("Could not run the subscription update process for keygroup "
+								+ keygroupID, e);
+					}
+				}
+
+			} catch (FBaseStorageConnectorException e1) {
+				logger.error(
+						"Could not read data to detect lost responsibilities, going back to sleep",
+						e1);
+			}
+
+			try {
+				Thread.sleep(checkInterval);
+			} catch (InterruptedException e) {
+				Thread.currentThread().interrupt();
+				logger.error("Background task has been interrupted");
+			}
+		}
+		logger.info("Stopping task, because Thread was interrupted");
+		return true;
+	}
+
+}

src/main/java/tasks/background/DetectMissingResponsibility.java

@@ -68,7 +68,7 @@ public Boolean executeFunctionality() {
 				}
 
 			} catch (FBaseStorageConnectorException e1) {
-				logger.error("Could not read heartbeats, going back to sleep", e1);
+				logger.error("Could not read responsibilities, going back to sleep", e1);
 			}
 
 			try {