Merge dev branch into master branch

.gitignore

@@ -28,3 +28,7 @@ temp/
 resources/test/
 demo/ScalaDemo/.DS_Store
 dependency-reduced-pom.xml
+test/
+.vscode/
+.metals/
+.bloop/

demo/DataLoader/pom.xml

@@ -161,21 +161,6 @@
             <artifactId>scala-logging_${scala.binary.version}</artifactId>
         </dependency>
 
-        <dependency>
-            <groupId>org.slf4j</groupId>
-            <artifactId>slf4j-api</artifactId>
-        </dependency>
-
-        <dependency>
-            <groupId>ch.qos.logback</groupId>
-            <artifactId>logback-core</artifactId>
-        </dependency>
-
-        <dependency>
-            <groupId>ch.qos.logback</groupId>
-            <artifactId>logback-classic</artifactId>
-        </dependency>
-
         <dependency>
             <groupId>com.typesafe.akka</groupId>
             <artifactId>akka-actor_${scala.binary.version}</artifactId>
@@ -216,13 +201,6 @@
             <artifactId>ganymed-ssh2</artifactId>
         </dependency>
 
-        <!-- https://mvnrepository.com/artifact/org.slf4j/slf4j-log4j12 -->
-        <dependency>
-            <groupId>org.slf4j</groupId>
-            <artifactId>slf4j-log4j12</artifactId>
-            <!-- <scope>test</scope> -->
-        </dependency>
-
         <dependency>
             <groupId>org.apache.spark</groupId>
             <artifactId>spark-core_${scala.binary.version}</artifactId>

demo/DataLoader/src/main/resources/log4j2.properties

@@ -0,0 +1,12 @@
+# Console logger
+appender.console.type=Console
+appender.console.name=STDOUT
+appender.console.layout.type=PatternLayout
+appender.console.layout.pattern=%d{yy/MM/dd HH:mm:ss} %p %c{1}: %m%n
+appender.console.filter.threshold.type=ThresholdFilter
+appender.console.filter.threshold.level=DEBUG
+
+# configure logger
+rootLogger=INFO,STDOUT
+
+

demo/DataLoader/src/main/scala/com/hackerforfuture/codeprototypes/dataloader/DataLoader.scala

@@ -31,20 +31,22 @@ object DataLoader extends LogSupport {
         }
       } catch {
         case e: ReflectiveOperationException =>
-          log.warn("Failed to register optional signal handler that logs a message when the process is terminated " +
-            s"by a signal. Reason for registration failure is: $e", e)
+          logger.warn("Failed to register optional signal handler that logs a message " +
+            "when the process is terminated by a signal. " +
+            "Reason for registration failure is: $e", e)
       }
 
       // attach shutdown handler to catch terminating signals as well as normal termination
       Runtime.getRuntime.addShutdownHook(new Thread("dataloader-shutdown-hook") {
         override def run(): Unit = DataLoaderServer.shutdown()
       })
 
+      logger.info("Starting DataLoaderServer...")
       DataLoaderServer.startup()
       DataLoaderServer.awaitShutdown()
     } catch {
       case NonFatal(e) =>
-        log.error("Failed to run DataLoader", e)
+        logger.error("Failed to run DataLoader", e)
         System.exit(1)
     }
     System.exit(0)

demo/DataLoader/src/main/scala/com/hackerforfuture/codeprototypes/dataloader/clusters/EventHandler.scala

@@ -0,0 +1,19 @@
+package com.hackerforfuture.codeprototypes.dataloader.clusters
+
+/**
+ * Author: biyu.huang
+ * Date: 2023/11/2 11:30
+ * Description:
+ */
+trait EventHandler {
+
+  def handleHeartbeatEvent(): Unit
+
+  def handleRegisterEvent(): Unit
+
+  def handleStopEvent(): Unit
+
+  def handleRegisterTimeout(): Unit
+
+  def handelCheckHeartbeatEvent(): Unit
+}

demo/DataLoader/src/main/scala/com/hackerforfuture/codeprototypes/dataloader/clusters/Message.scala

@@ -0,0 +1,25 @@
+package com.hackerforfuture.codeprototypes.dataloader.clusters
+
+import akka.actor.ActorPath
+
+/**
+ * Author: biyu.huang
+ * Date: 2023/11/1 19:06
+ * Description:
+ */
+sealed trait Message
+
+case object Heartbeat extends Message
+
+case object CheckHeartbeat extends Message
+
+case object Register extends Message
+
+case object RegisterTimeout extends Message
+
+// 自定义消息类型，带有唯一标识ID
+case class CustomMessage(id: ActorPath, content: Any) extends Message
+
+case class SlaveActorTerminated(id: ActorPath, reason: String) extends Message
+
+case object StopActor extends Message

demo/DataLoader/src/main/scala/com/hackerforfuture/codeprototypes/dataloader/clusters/master/MasterActor.scala

@@ -0,0 +1,105 @@
+package com.hackerforfuture.codeprototypes.dataloader.clusters.master
+
+import akka.actor.{Actor, ActorLogging, ActorPath, Props, Terminated}
+import com.hackerforfuture.codeprototypes.dataloader.clusters._
+import com.hackerforfuture.codeprototypes.dataloader.common.Using
+
+import scala.collection.mutable
+import scala.concurrent.duration._
+
+/**
+ * Author: biyu.huang
+ * Date: 2023/11/1 17:15
+ * Description:
+ */
+object MasterActor {
+  def props: Props = Props[MasterActor]
+}
+
+class MasterActor extends Actor with ActorLogging with EventHandler with Using {
+  // 存储已注册的SlaveActor的唯一标识ID和最后一次收到心跳的时间戳
+  private final val registeredSlaves = mutable.HashMap.empty[ActorPath, Long]
+
+  // 心跳超时时间
+  private final val heartbeatTimeout: FiniteDuration = 10.seconds
+
+  private final val initialTimeout: FiniteDuration = 5.seconds
+
+  // 启动时设置定时器
+  override def preStart(): Unit = {
+    //    context.setReceiveTimeout(initialTimeout)
+
+    import context.dispatcher
+    context
+      .system
+      .scheduler
+      .schedule(initialTimeout, heartbeatTimeout, self, CheckHeartbeat)
+  }
+
+  override def handleHeartbeatEvent(): Unit = {
+    syncableBlock {
+      val senderId: ActorPath = sender().path
+      val ts: Long = System.currentTimeMillis()
+      if (registeredSlaves.contains(senderId)) {
+        registeredSlaves.put(senderId, ts)
+        log.info(s"[%s] Received heartbeat from slave: ${senderId.name}".format(ts))
+      } else {
+        registeredSlaves.put(senderId, ts)
+        log.info(s"[%s] Registered slave: ${senderId.name}".format(ts))
+        context.watch(sender())
+      }
+    }
+  }
+
+  override def handleRegisterEvent(): Unit = handleHeartbeatEvent()
+
+  override def handelCheckHeartbeatEvent(): Unit = {
+    val currentTime: Long = System.currentTimeMillis()
+    val timedOutSlaves = registeredSlaves.filter {
+      case (_, lastHeartbeatTime) =>
+        val elapsed = currentTime - lastHeartbeatTime
+        elapsed > (heartbeatTimeout.toMillis * 10)
+    }
+
+    timedOutSlaves.keys.foreach { id =>
+      registeredSlaves -= id
+      log.warning(s"Slave $id timed out and unregistered.")
+      self ! SlaveActorTerminated(id, "No heartbeat received")
+    }
+  }
+
+  override def handleStopEvent(): Unit = {
+    log.info("received StopActor message, shutting down ...")
+    registeredSlaves.foreach {
+      case (id, _) =>
+        log.info("try to stop %s".format(id.name))
+        context.system.actorSelection(id) ! StopActor
+    }
+    registeredSlaves.clear()
+    context.stop(self)
+  }
+
+  def receive: Receive = {
+    case Heartbeat => handleHeartbeatEvent()
+    case Register => handleRegisterEvent()
+    case CheckHeartbeat => handelCheckHeartbeatEvent()
+    case StopActor => handleStopEvent()
+    case CustomMessage(id, content) =>
+      if (registeredSlaves.contains(id)) {
+        log.info(s"Processing message from slave $id: $content")
+        // 处理消息逻辑
+      } else {
+        log.warning(s"Received message from unregistered slave: $id")
+        context.system.actorSelection(id) ! RegisterTimeout
+      }
+    case SlaveActorTerminated(id, reason) =>
+      log.warning(s"Slave $id terminated and unregistered. Reason: $reason")
+      registeredSlaves.remove(id)
+    case Terminated(slave) =>
+      val slaveId = slave.path
+      registeredSlaves.remove(slaveId)
+      log.warning(s"Slave ${slaveId.name} terminated.")
+  }
+
+  override def handleRegisterTimeout(): Unit = {}
+}