Fix HCS latency regression (0.97)

Fix the HCS E2E latency regression by invoking RedisPublisher before SqlEntityListener

Signed-off-by: Steven Sheehy <[email protected]>

hedera-mirror-importer/src/main/java/com/hedera/mirror/importer/parser/record/entity/notify/NotifyingPublisher.java

@@ -37,7 +37,7 @@
 @ConditionOnEntityRecordParser
 @CustomLog
 @Named
-@Order(2)
+@Order(1)
 public class NotifyingPublisher implements BatchPublisher {
 
     private static final String SQL = "select pg_notify('topic_message', ?)";

hedera-mirror-importer/src/main/java/com/hedera/mirror/importer/parser/record/entity/redis/RedisPublisher.java

@@ -42,7 +42,7 @@
 @ConditionOnEntityRecordParser
 @CustomLog
 @Named
-@Order(1)
+@Order(0) // Triggering the async publishing before other operations can reduce latency
 public class RedisPublisher implements BatchPublisher {
 
     private static final String TOPIC_FORMAT = "topic.%d";

hedera-mirror-importer/src/main/java/com/hedera/mirror/importer/parser/record/entity/sql/SqlEntityListener.java

@@ -75,7 +75,7 @@
 
 @CustomLog
 @Named
-@Order(0)
+@Order(2)
 @ConditionOnEntityRecordParser
 @RequiredArgsConstructor
 public class SqlEntityListener implements EntityListener, RecordStreamFileListener {

hedera-mirror-importer/src/main/java/com/hedera/mirror/importer/parser/record/entity/topic/TopicMessageLookupEntityListener.java

@@ -37,7 +37,7 @@
 
 @CustomLog
 @Named
-@Order(1)
+@Order(3)
 @ConditionOnEntityRecordParser
 public class TopicMessageLookupEntityListener implements EntityListener, RecordStreamFileListener {
 

hedera-mirror-importer/src/test/java/com/hedera/mirror/importer/parser/record/entity/redis/RedisPublisherIntegrationTest.java

@@ -16,11 +16,16 @@
 
 package com.hedera.mirror.importer.parser.record.entity.redis;
 
+import static org.assertj.core.api.Assertions.assertThat;
+
 import com.hedera.mirror.common.domain.entity.EntityId;
 import com.hedera.mirror.common.domain.topic.StreamMessage;
 import com.hedera.mirror.common.domain.topic.TopicMessage;
+import com.hedera.mirror.importer.parser.record.RecordStreamFileListener;
 import com.hedera.mirror.importer.parser.record.entity.BatchPublisherTest;
 import com.hedera.mirror.importer.parser.record.entity.ParserContext;
+import java.util.List;
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInstance;
 import org.springframework.data.redis.core.ReactiveRedisOperations;
 import reactor.core.publisher.Flux;
@@ -29,18 +34,26 @@
 class RedisPublisherIntegrationTest extends BatchPublisherTest {
 
     private final ReactiveRedisOperations<String, StreamMessage> redisOperations;
+    private final List<RecordStreamFileListener> streamFileListeners;
 
     public RedisPublisherIntegrationTest(
-            RedisPublisher entityListener,
+            RedisPublisher redisPublisher,
             ParserContext parserContext,
             RedisProperties properties,
-            ReactiveRedisOperations<String, StreamMessage> redisOperations) {
-        super(entityListener, parserContext, properties);
+            ReactiveRedisOperations<String, StreamMessage> redisOperations,
+            List<RecordStreamFileListener> streamFileListeners) {
+        super(redisPublisher, parserContext, properties);
         this.redisOperations = redisOperations;
+        this.streamFileListeners = streamFileListeners;
     }
 
     @Override
     protected Flux<TopicMessage> subscribe(EntityId topicId) {
         return redisOperations.listenToChannel("topic." + topicId.getId()).map(m -> (TopicMessage) m.getMessage());
     }
+
+    @Test
+    void publishesFirst() {
+        assertThat(streamFileListeners).first().isEqualTo(batchPublisher);
+    }
 }