From fc3771c5bf4cf309826b9ba3c2f32204a667f24b Mon Sep 17 00:00:00 2001
From: patschuh <46817726+patschuh@users.noreply.github.com>
Date: Fri, 7 Jun 2024 21:00:41 +0200
Subject: [PATCH] #46 Attempt to reduce jvm crashes on macOS

---
 src/main/java/at/esque/kafka/Controller.java | 46 +++++++++++++++-----
 1 file changed, 34 insertions(+), 12 deletions(-)

diff --git a/src/main/java/at/esque/kafka/Controller.java b/src/main/java/at/esque/kafka/Controller.java
index 69f4ad2..18199bf 100644
--- a/src/main/java/at/esque/kafka/Controller.java
+++ b/src/main/java/at/esque/kafka/Controller.java
@@ -591,7 +591,7 @@ private void getTopicsForCluster() {
         } finally {
             stopWatch.stop();
             LOGGER.info("Finished getting topics for cluster [{}]", stopWatch);
-            backGroundTaskHolder.backgroundTaskStopped();
+            Platform.runLater(() -> backGroundTaskHolder.backgroundTaskStopped());
         }
     }
 
@@ -825,8 +825,10 @@ private void getOldestMessages(TopicMessageTypeConfig topic, Map<String, String>
             }
             try {
                 Map<Integer, AtomicLong> messagesConsumed = new HashMap<>();
-                Platform.runLater(() -> backGroundTaskHolder.setBackGroundTaskDescription("preparing consumer..."));
-                backGroundTaskHolder.setIsInProgress(true);
+                Platform.runLater(() -> {
+                    backGroundTaskHolder.setBackGroundTaskDescription("preparing consumer...");
+                    backGroundTaskHolder.setIsInProgress(true);
+                });
                 subscribeOrAssignToSelectedPartition(topic, consumerId);
                 Map<TopicPartition, Long> minOffsets = consumerHandler.getMinOffsets(consumerId);
                 Map<TopicPartition, Long> maxOffsets = consumerHandler.getMaxOffsets(consumerId);
@@ -853,7 +855,15 @@ private ObservableList<KafkaMessage> getAndClearBaseList(PinTab tab) {
             return null;
         }
         ObservableList<KafkaMessage> baseList = ((MessagesTabContent) tab.getContent()).getMessageTableView().getBaseList();
-        baseList.clear();
+        Platform.runLater(baseList::clear);
+        while (!baseList.isEmpty()) {
+            LOGGER.info("Waiting for baseList to be cleared by FX thread");
+            try {
+                Thread.sleep(100);
+            } catch (InterruptedException e) {
+                throw new RuntimeException(e);
+            }
+        }
         return baseList;
     }
 
@@ -900,8 +910,11 @@ private void getNewestMessages(TopicMessageTypeConfig topic, Map<String, String>
             UUID consumerId = tempconsumerId;
             try {
                 Map<Integer, AtomicLong> messagesConsumed = new HashMap<>();
-                Platform.runLater(() -> backGroundTaskHolder.setBackGroundTaskDescription("preparing consumer..."));
-                backGroundTaskHolder.setIsInProgress(true);
+                Platform.runLater(() -> {
+                            backGroundTaskHolder.setBackGroundTaskDescription("preparing consumer...");
+                            backGroundTaskHolder.setIsInProgress(true);
+                        }
+                );
                 subscribeOrAssignToSelectedPartition(topic, consumerId);
                 Map<TopicPartition, Long> minOffsets = consumerHandler.getMinOffsets(consumerId);
                 Map<TopicPartition, Long> maxOffsets = consumerHandler.getMaxOffsets(consumerId);
@@ -963,7 +976,7 @@ private <KT, VT> void getMessagesContinuously(TopicMessageTypeConfig topic, Map<
         runInDaemonThread(() -> {
             try {
                 AtomicLong messagesConsumed = new AtomicLong(0);
-                backGroundTaskHolder.setIsInProgress(true);
+                Platform.runLater(() -> backGroundTaskHolder.setIsInProgress(true));
                 subscribeOrAssignToSelectedPartition(topic, consumerId);
                 consumerHandler.seekToOffset(consumerId, -2);
                 PinTab tab = getActiveTabOrAddNew(topic, false);
@@ -995,7 +1008,7 @@ private <KT, VT> void trace(TopicMessageTypeConfig topic, Map<String, String> co
                 return;
             }
             try {
-                backGroundTaskHolder.setIsInProgress(true);
+                Platform.runLater(() -> backGroundTaskHolder.setIsInProgress(true));
                 List<TopicPartition> topicPatitions;
                 if (fasttracePartition != null) {
                     topicPatitions = new ArrayList<>(Collections.singletonList(new TopicPartition(selectedTopic(), fasttracePartition)));
@@ -1135,8 +1148,11 @@ private void getMessagesFromSpecificOffset(TopicMessageTypeConfig topic, Map<Str
             try {
                 Map<Integer, AtomicLong> messagesConsumed = new HashMap<>();
                 long specifiedOffset = Long.parseLong(specificOffsetTextField.getText());
-                Platform.runLater(() -> backGroundTaskHolder.setBackGroundTaskDescription("preparing consumer..."));
-                backGroundTaskHolder.setIsInProgress(true);
+                Platform.runLater(() -> {
+                            backGroundTaskHolder.setBackGroundTaskDescription("preparing consumer...");
+                            backGroundTaskHolder.setIsInProgress(true);
+                        }
+                );
                 subscribeOrAssignToSelectedPartition(topic, consumerId);
                 Map<TopicPartition, Long> minOffsets = consumerHandler.getMinOffsets(consumerId);
                 Map<TopicPartition, Long> maxOffsets = consumerHandler.getMaxOffsets(consumerId);
@@ -1172,8 +1188,11 @@ private void getMessagesStartingFromInstant(TopicMessageTypeConfig topic, Map<St
             }
             try {
                 Map<Integer, AtomicLong> messagesConsumed = new HashMap<>();
-                Platform.runLater(() -> backGroundTaskHolder.setBackGroundTaskDescription("preparing consumer..."));
-                backGroundTaskHolder.setIsInProgress(true);
+                Platform.runLater(() -> {
+                            backGroundTaskHolder.setBackGroundTaskDescription("preparing consumer...");
+                            backGroundTaskHolder.setIsInProgress(true);
+                        }
+                );
                 subscribeOrAssignToSelectedPartition(topic, consumerId);
                 Map<TopicPartition, Long> minOffsets = consumerHandler.getMinOffsets(consumerId);
                 Map<TopicPartition, Long> maxOffsets = consumerHandler.getMaxOffsets(consumerId);
@@ -1332,6 +1351,9 @@ private void runInDaemonThread(Runnable runnable) {
         Thread daemonThread = new Thread(runnable);
         daemonThread.setDaemon(true);
         daemonThread.start();
+        daemonThread.setUncaughtExceptionHandler((t, e) -> {
+            Platform.runLater(() -> ErrorAlert.show(e, controlledStage));
+        });
     }
 
     // Experimental Area