remove test

src/main/scala/com.gmei.up/Main.scala

@@ -45,9 +45,8 @@ object Main extends LazyLogging {
         "172.16.44.25:9092,172.16.44.31:9092,172.16.44.45:9092"
       )
 
-      // TODO remove test
       val kafkaConsumer = new FlinkKafkaConsumer[UserInfo](
-        "gm-portrait-update-device-test",
+        "gm-portrait-update-device",
         new UserInfoDeserializationSchema,
         kafkaConsumerProperties
       )

src/main/scala/com.gmei.up/utils/Redis.scala

@@ -34,34 +34,47 @@ object Redis {
   ): Unit = {
     val key = s"streaming:candidate:${contentType}:device_id:${deviceId}"
 
-    contentEitherFuture match {
-      case Left(e) => e.printStackTrace()
-      case Right(contentFuture) =>
-        pRc4Pool.withClient { client =>
-          contentFuture.foreach { seq =>
-            val ids = seq.map(c => c.id)
-            if (ids.size > 0) {
-              client.del(key)
-              ids.foreach { id =>
-                // println(id)
-                client.rpush(key, id)
-              }
-              client.expire(key, 60 * 60 * 24 * 15)
-              val total = System.currentTimeMillis - timeBegin
-              if (total > 1000 && total < 50000) {
-                DingTalk.send(
-                  Map(
-                    "method" -> "Redis.save",
-                    "deviceId" -> deviceId,
-                    "contentType" -> contentType,
-                    "description" -> "从流开始处理到存入结果的总耗时",
-                    "streamTotalSeconds" -> s"${total}ms"
+    try {
+      contentEitherFuture match {
+        case Left(e) => e.printStackTrace()
+        case Right(contentFuture) =>
+          pRc4Pool.withClient { client =>
+            contentFuture.foreach { seq =>
+              val ids = seq.map(c => c.id)
+              if (ids.size > 0) {
+                client.del(key)
+                ids.foreach { id =>
+                  // println(id)
+                  client.rpush(key, id)
+                }
+                client.expire(key, 60 * 60 * 24 * 15)
+                val total = System.currentTimeMillis - timeBegin
+                if (total > 1000 && total < 50000) {
+                  DingTalk.send(
+                    Map(
+                      "method" -> "Redis.save",
+                      "deviceId" -> deviceId,
+                      "contentType" -> contentType,
+                      "description" -> "从流开始处理到存入结果的总耗时",
+                      "streamTotalSeconds" -> s"${total}ms"
+                    )
                   )
-                )
+                }
               }
             }
           }
-        }
+      }
+    } catch {
+      case e: Throwable =>
+        DingTalk.send(
+          Map(
+            "method" -> "Redis.save",
+            "deviceId" -> deviceId,
+            "contentType" -> contentType,
+            "error" -> e.getStackTrace.mkString("\n")
+          ),
+          contentType = "exception"
+        )
     }
   }
 
@@ -88,7 +101,12 @@ object Redis {
     } catch {
       case e: Throwable =>
         DingTalk.send(
-          Map("method" -> "Redis.saveServiceDiary", "error" -> e.getStackTrace.mkString("\n")),
+          Map(
+            "method" -> "Redis.saveServiceDiary",
+            "deviceId" -> deviceId,
+            "contentType" -> "service_diary",
+            "error" -> e.getStackTrace.mkString("\n")
+          ),
           contentType = "exception"
         )
     }