From db1538ae32d9099255e1e90946c0bfb2b4e59d55 Mon Sep 17 00:00:00 2001
From: "chenglong.wang" <tiny.wcl@gmail.com>
Date: Tue, 14 Nov 2023 11:30:49 +0800
Subject: [PATCH] improve: Log invalid edges

---
 .../janusgraph/internal/executor/JanusGraphEdgeExecutor.java | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/src/main/java/org/apache/flink/connector/janusgraph/internal/executor/JanusGraphEdgeExecutor.java b/src/main/java/org/apache/flink/connector/janusgraph/internal/executor/JanusGraphEdgeExecutor.java
index 338727e..971ef08 100644
--- a/src/main/java/org/apache/flink/connector/janusgraph/internal/executor/JanusGraphEdgeExecutor.java
+++ b/src/main/java/org/apache/flink/connector/janusgraph/internal/executor/JanusGraphEdgeExecutor.java
@@ -105,6 +105,11 @@ protected void execute(RowData record, JanusGraphTransaction transaction) {
     private void createEdge(Object[] values, JanusGraphTransaction transaction) {
         Vertex inV = inVertexSearcher.search(values, transaction);
         Vertex outV = outVertexSearcher.search(values, transaction);
+        if (inV == null || outV == null) {
+            LOG.debug("in vertex: {} and out vertex: {} must be present", inV, outV);
+            return;
+        }
+
         Edge created = outV.addEdge(values[labelIndex].toString(), inV);
         for (int i = 0; i < values.length; i++) {
             if (!nonWriteColumnIndexes.contains(i)) {