Log dataflow messages at appropriate levels (#27788)

* Log dataflow messages at appropriate levels

* Use single-cast message importance for error case

* Add catch-all

sdks/python/apache_beam/runners/dataflow/dataflow_runner.py

@@ -197,11 +197,21 @@ def rank_error(msg):
           # Skip empty messages.
           if m.messageImportance is None:
             continue
-          _LOGGER.info(message)
-          if str(m.messageImportance) == 'JOB_MESSAGE_ERROR':
+          message_importance = str(m.messageImportance)
+          if (message_importance == 'JOB_MESSAGE_DEBUG' or
+              message_importance == 'JOB_MESSAGE_DETAILED'):
+            _LOGGER.debug(message)
+          elif message_importance == 'JOB_MESSAGE_BASIC':
+            _LOGGER.info(message)
+          elif message_importance == 'JOB_MESSAGE_WARNING':
+            _LOGGER.warning(message)
+          elif message_importance == 'JOB_MESSAGE_ERROR':
+            _LOGGER.error(message)
             if rank_error(m.messageText) >= last_error_rank:
               last_error_rank = rank_error(m.messageText)
               last_error_msg = m.messageText
+          else:
+            _LOGGER.info(message)
         if not page_token:
           break