Lint code

Author: reembs

src/main/kotlin/org/xyro/kumulus/KumulusTopology.kt

@@ -301,12 +301,13 @@ class KumulusTopology(
             logger.trace { "Component ${c.componentId}/${c.taskId} is currently busy" }
             var shouldEnqueue = true
             if (message is ExecuteMessage) {
-                val messageWaitStartTime =  c.waitStart.get()
-                if (messageWaitStartTime > 0){
+                val messageWaitStartTime = c.waitStart.get()
+                if (messageWaitStartTime > 0) {
                     val delay = System.nanoTime() - messageWaitStartTime
                     if ((delay >= lateMessageMaxWaitInNanos) &&
-                        lateMessagesStreamsToDrop.contains(message.tuple.kTuple.sourceStreamId)) {
-                        if (!message.isLate.get()){
+                        lateMessagesStreamsToDrop.contains(message.tuple.kTuple.sourceStreamId)
+                    ) {
+                        if (!message.isLate.get()) {
                             message.isLate.set(true)
                             onLateMessageHook?.let { onLateMessageHook ->
                                 try {
@@ -328,7 +329,7 @@ class KumulusTopology(
                             }
                         }
 
-                        if (lateMessagesShouldDrop){
+                        if (lateMessagesShouldDrop) {
                             shouldEnqueue = false
                         }
                     }
@@ -337,7 +338,7 @@ class KumulusTopology(
                 c.waitStart.compareAndSet(0, System.nanoTime())
             }
 
-            if (shouldEnqueue){
+            if (shouldEnqueue) {
                 if (queuePushbackWait <= 0L) {
                     boltExecutionPool.enqueue(message)
                 } else {

src/test/kotlin/org/xyro/kumulus/TestDroppingStaleMessages.kt

@@ -22,9 +22,9 @@ class TestDroppingStaleMessages {
         val config: MutableMap<String, Any> = mutableMapOf()
         config[Config.TOPOLOGY_MAX_SPOUT_PENDING] = 1L
         config[KumulusTopology.CONF_THREAD_POOL_CORE_SIZE] = 5L
-        config[KumulusTopology.CONF_LATE_MESSAGES_DROPPING_SHOULD_DROP] = true;
-        config[KumulusTopology.CONF_LATE_MESSAGES_DROPPING_STREAMS_NAME] = setOf("default");
-        config[KumulusTopology.CONF_LATE_MESSAGES_DROPPING_MAX_WAIT_SECONDS] = 1;
+        config[KumulusTopology.CONF_LATE_MESSAGES_DROPPING_SHOULD_DROP] = true
+        config[KumulusTopology.CONF_LATE_MESSAGES_DROPPING_STREAMS_NAME] = setOf("default")
+        config[KumulusTopology.CONF_LATE_MESSAGES_DROPPING_MAX_WAIT_SECONDS] = 1
 
         builder.setSpout("spout", LatencyDeltaSpout())
 
@@ -34,7 +34,6 @@ class TestDroppingStaleMessages {
         builder.setBolt("delay-unanchored-bolt", StuckBolt())
             .noneGrouping("unanchoring-bolt")
 
-
         val stormTopology = builder.createTopology()!!
         val kumulusTopology =
             KumulusStormTransformer.initializeTopology(stormTopology, config, "test")
@@ -50,8 +49,8 @@ class TestDroppingStaleMessages {
         Thread.sleep(5000)
         kumulusTopology.stop()
 
-        logger.info { "Dropped ${lateHookCalled} messages" }
-        assertTrue { lateHookCalled}
+        logger.info { "Dropped $lateHookCalled messages" }
+        assertTrue { lateHookCalled }
     }
 
     class LatencyDeltaSpout : DummySpout({
@@ -110,7 +109,7 @@ class TestDroppingStaleMessages {
     }) {
         override fun execute(input: Tuple, collector: BasicOutputCollector) {
             logger.info { "StuckBolt: started" }
-            while (true){
+            while (true) {
                 Thread.sleep(50)
             }
         }