diff --git a/dp3/task_processing/task_queue.py b/dp3/task_processing/task_queue.py
index 9f99d443..d7a58687 100644
--- a/dp3/task_processing/task_queue.py
+++ b/dp3/task_processing/task_queue.py
@@ -538,7 +538,10 @@ def _msg_processing_thread_func(self):
                 continue
 
             # Pass message to user's callback function
-            self.callback(tag, task)
+            try:
+                self.callback(tag, task)
+            except Exception as e:
+                self.log.exception("Error in user callback function. %s: %s", type(e), str(e))
 
     def _stop_consuming_thread(self) -> None:
         if self._consuming_thread: