fix(spark-lineage): default timeout for future responses

metadata-integration/java/datahub-client/src/main/java/datahub/client/MetadataResponseFuture.java

@@ -69,7 +69,7 @@ public MetadataWriteResponse get(long timeout, TimeUnit unit)
       return mapper.map(response);
     } else {
       // We wait for the callback to fill this out
-      responseLatch.await();
+      responseLatch.await(timeout, unit);
       return responseReference.get();
     }
   }

metadata-integration/java/spark-lineage-beta/src/main/java/datahub/spark/DatahubEventEmitter.java

@@ -39,6 +39,8 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
@@ -53,6 +55,7 @@ public class DatahubEventEmitter extends EventEmitter {
   private final List<DatahubJob> _datahubJobs = new LinkedList<>();
   private final Map<String, MetadataChangeProposalWrapper> schemaMap = new HashMap<>();
   private SparkLineageConf datahubConf;
+  private static final int DEFAULT_TIMEOUT_SEC = 10;
 
   private final EventFormatter eventFormatter = new EventFormatter();
 
@@ -386,8 +389,8 @@ protected void emitMcps(List<MetadataChangeProposal> mcps) {
           .forEach(
               future -> {
                 try {
-                  log.info(future.get().toString());
-                } catch (InterruptedException | ExecutionException e) {
+                  log.info(future.get(DEFAULT_TIMEOUT_SEC, TimeUnit.SECONDS).toString());
+                } catch (InterruptedException | ExecutionException | TimeoutException e) {
                   // log error, but don't impact thread
                   log.error("Failed to emit metadata to DataHub", e);
                 }