Handle invalid entity ID in GCP Pub/Sub (#3257)

Signed-off-by: Steven Sheehy <steven.sheehy@hedera.com> Signed-off-by: Matheus DallRosa <matheus.dallrosa@swirlds.com>
hashgraph · Feb 21, 2022 · 8b79cf0 · 8b79cf0
1 parent a136da8
commit 8b79cf0
Show file tree

Hide file tree

Showing 2 changed files with 36 additions and 8 deletions.
diff --git a/...c/main/java/com/hedera/mirror/importer/parser/record/pubsub/PubSubRecordItemListener.java b/...c/main/java/com/hedera/mirror/importer/parser/record/pubsub/PubSubRecordItemListener.java
@@ -20,8 +20,6 @@
  * ‍
  */
 
-import com.hedera.mirror.common.util.DomainUtils;
-
 import com.hederahashgraph.api.proto.java.AccountAmount;
 import com.hederahashgraph.api.proto.java.FileID;
 import com.hederahashgraph.api.proto.java.TransactionBody;
@@ -37,6 +35,8 @@
 import com.hedera.mirror.common.domain.file.FileData;
 import com.hedera.mirror.common.domain.transaction.RecordItem;
 import com.hedera.mirror.common.domain.transaction.TransactionType;
+import com.hedera.mirror.common.exception.InvalidEntityException;
+import com.hedera.mirror.common.util.DomainUtils;
 import com.hedera.mirror.importer.addressbook.AddressBookService;
 import com.hedera.mirror.importer.exception.ImporterException;
 import com.hedera.mirror.importer.exception.ParserException;
@@ -69,8 +69,16 @@ public void onItem(RecordItem recordItem) throws ImporterException {
         TransactionHandler transactionHandler = transactionHandlerFactory.get(transactionType);
         log.trace("Storing transaction body: {}", () -> Utility.printProtoMessage(body));
         long consensusTimestamp = DomainUtils.timeStampInNanos(txRecord.getConsensusTimestamp());
-        EntityId entity = transactionHandler.getEntity(recordItem);
-        PubSubMessage pubSubMessage = buildPubSubMessage(consensusTimestamp, entity, recordItem);
+
+        EntityId entityId;
+        try {
+            entityId = transactionHandler.getEntity(recordItem);
+        } catch (InvalidEntityException e) { // transaction can have invalid topic/contract/file id
+            log.warn("Invalid entity encountered for consensusTimestamp {} : {}", consensusTimestamp, e.getMessage());
+            entityId = null;
+        }
+
+        PubSubMessage pubSubMessage = buildPubSubMessage(consensusTimestamp, entityId, recordItem);
         try {
             sendPubSubMessage(pubSubMessage);
         } catch (Exception e) {
@@ -81,7 +89,7 @@ public void onItem(RecordItem recordItem) throws ImporterException {
         }
         log.debug("Published transaction : {}", consensusTimestamp);
 
-        if (addressBookService.isAddressBook(entity)) {
+        if (addressBookService.isAddressBook(entityId)) {
             FileID fileID = null;
             byte[] fileBytes = null;
 
@@ -137,4 +145,3 @@ private Iterable<AccountAmount> addNonFeeTransfers(TransactionBody body, Transac
         return nonFeeTransfers;
     }
 }
-
diff --git a/...st/java/com/hedera/mirror/importer/parser/record/pubsub/PubSubRecordItemListenerTest.java b/...st/java/com/hedera/mirror/importer/parser/record/pubsub/PubSubRecordItemListenerTest.java
@@ -65,11 +65,11 @@
 
 import com.hedera.mirror.common.domain.entity.EntityId;
 import com.hedera.mirror.common.domain.file.FileData;
-import com.hedera.mirror.importer.addressbook.AddressBookService;
+import com.hedera.mirror.common.domain.transaction.RecordItem;
 import com.hedera.mirror.common.domain.transaction.TransactionType;
+import com.hedera.mirror.importer.addressbook.AddressBookService;
 import com.hedera.mirror.importer.exception.ParserException;
 import com.hedera.mirror.importer.parser.domain.PubSubMessage;
-import com.hedera.mirror.common.domain.transaction.RecordItem;
 import com.hedera.mirror.importer.parser.record.NonFeeTransferExtractionStrategy;
 import com.hedera.mirror.importer.parser.record.NonFeeTransferExtractionStrategyImpl;
 import com.hedera.mirror.importer.parser.record.transactionhandler.TransactionHandler;
@@ -170,6 +170,27 @@ void testPubSubMessage() throws Exception {
         assertThat(pubSubMessage.getNonFeeTransfers()).isNull();
     }
 
+    @Test
+    void testPubSubMessageNullEntityId() throws Exception {
+        // given
+        byte[] message = new byte[] {'a', 'b', 'c'};
+        TopicID topicID = TopicID.newBuilder().setTopicNum(10L).build();
+        EntityId topicIdEntity = EntityId.of(topicID);
+        ConsensusSubmitMessageTransactionBody submitMessage = ConsensusSubmitMessageTransactionBody.newBuilder()
+                .setMessage(ByteString.copyFrom(message))
+                .setTopicID(topicID)
+                .build();
+        Transaction transaction = buildTransaction(builder -> builder.setConsensusSubmitMessage(submitMessage));
+        // when
+        doReturn(null).when(transactionHandler).getEntity(any());
+        pubSubRecordItemListener.onItem(new RecordItem(transaction.toByteArray(), DEFAULT_RECORD_BYTES));
+
+        // then
+        var pubSubMessage = assertPubSubMessage(buildPubSubTransaction(transaction), 1);
+        assertThat(pubSubMessage.getEntity()).isEqualTo(null);
+        assertThat(pubSubMessage.getNonFeeTransfers()).isNull();
+    }
+
     @Test
     void testPubSubMessageWithNonFeeTransferAndNullEntityId() throws Exception {
         // given