diff --git a/google/cloud/logging_v2/client.py b/google/cloud/logging_v2/client.py
index 92ab72a3a..3d5ea24fc 100644
--- a/google/cloud/logging_v2/client.py
+++ b/google/cloud/logging_v2/client.py
@@ -376,7 +376,7 @@ def get_default_handler(self, **kw):
             if monitored_resource.type == _GAE_RESOURCE_TYPE:
                 return CloudLoggingHandler(self, resource=monitored_resource, **kw)
             elif monitored_resource.type == _GKE_RESOURCE_TYPE:
-                return ContainerEngineHandler(**kw)
+                return StructuredLogHandler(**kw, project_id=self.project)
             elif monitored_resource.type == _GCF_RESOURCE_TYPE:
                 # __stdout__ stream required to support structured logging on Python 3.7
                 kw["stream"] = kw.get("stream", sys.__stdout__)
diff --git a/google/cloud/logging_v2/handlers/handlers.py b/google/cloud/logging_v2/handlers/handlers.py
index 8d14852e1..39bcbca79 100644
--- a/google/cloud/logging_v2/handlers/handlers.py
+++ b/google/cloud/logging_v2/handlers/handlers.py
@@ -221,9 +221,16 @@ def _format_and_parse_message(record, formatter_handler):
         record (logging.LogRecord): The record object representing the log
         formatter_handler (logging.Handler): The handler used to format the log
     """
-    # if message is a dictionary, return as-is
+    passed_json_fields = getattr(record, "json_fields", {})
+    # if message is a dictionary, use dictionary directly
     if isinstance(record.msg, collections.abc.Mapping):
-        return record.msg
+        payload = record.msg
+        # attach any extra json fields if present
+        if passed_json_fields and isinstance(
+            passed_json_fields, collections.abc.Mapping
+        ):
+            payload = {**payload, **passed_json_fields}
+        return payload
     # format message string based on superclass
     message = formatter_handler.format(record)
     try:
@@ -235,6 +242,11 @@ def _format_and_parse_message(record, formatter_handler):
     except (json.decoder.JSONDecodeError, IndexError):
         # log string is not valid json
         pass
+    # if json_fields was set, create a dictionary using that
+    if passed_json_fields and isinstance(passed_json_fields, collections.abc.Mapping):
+        if message != "None":
+            passed_json_fields["message"] = message
+        return passed_json_fields
     # if formatted message contains no content, return None
     return message if message != "None" else None
 
diff --git a/tests/environment b/tests/environment
index a201d48f1..2ff3cb26a 160000
--- a/tests/environment
+++ b/tests/environment
@@ -1 +1 @@
-Subproject commit a201d48f163db4ff799305188d380f620ebf345a
+Subproject commit 2ff3cb26a09d574715573bcf0a2c480eeac73311
diff --git a/tests/system/test_system.py b/tests/system/test_system.py
index cde722bd6..24050e8b3 100644
--- a/tests/system/test_system.py
+++ b/tests/system/test_system.py
@@ -551,6 +551,31 @@ def test_handlers_w_extras(self):
         )
         self.assertEqual(entries[0].resource.type, extra["resource"].type)
 
+    def test_handlers_w_json_fields(self):
+        LOG_MESSAGE = "Testing with json_field extras."
+        LOGGER_NAME = "json_field_extras"
+        handler_name = self._logger_name(LOGGER_NAME)
+
+        handler = CloudLoggingHandler(
+            Config.CLIENT, name=handler_name, transport=SyncTransport
+        )
+
+        # only create the logger to delete, hidden otherwise
+        logger = Config.CLIENT.logger(handler.name)
+        self.to_delete.append(logger)
+
+        cloud_logger = logging.getLogger(LOGGER_NAME)
+        cloud_logger.addHandler(handler)
+        extra = {"json_fields": {"hello": "world", "two": 2}}
+        cloud_logger.warn(LOG_MESSAGE, extra=extra)
+
+        entries = _list_entries(logger)
+        self.assertEqual(len(entries), 1)
+        payload = entries[0].payload
+        self.assertEqual(payload["message"], LOG_MESSAGE)
+        self.assertEqual(payload["hello"], "world")
+        self.assertEqual(payload["two"], 2)
+
     def test_log_root_handler(self):
         LOG_MESSAGE = "It was the best of times."
 
diff --git a/tests/unit/handlers/test_handlers.py b/tests/unit/handlers/test_handlers.py
index d36dc8959..71a709b6a 100644
--- a/tests/unit/handlers/test_handlers.py
+++ b/tests/unit/handlers/test_handlers.py
@@ -447,6 +447,40 @@ def test_emit_dict(self):
             ),
         )
 
+    def test_emit_w_json_extras(self):
+        """
+        User can add json_fields to the record, which should populate the payload
+        """
+        from google.cloud.logging_v2.logger import _GLOBAL_RESOURCE
+
+        client = _Client(self.PROJECT)
+        handler = self._make_one(
+            client, transport=_Transport, resource=_GLOBAL_RESOURCE,
+        )
+        message = "message"
+        json_fields = {"hello": "world"}
+        logname = "logname"
+        expected_label = {"python_logger": logname}
+        record = logging.LogRecord(
+            logname, logging.INFO, None, None, message, None, None
+        )
+        setattr(record, "json_fields", json_fields)
+        handler.handle(record)
+
+        self.assertEqual(
+            handler.transport.send_called_with,
+            (
+                record,
+                {"message": "message", "hello": "world"},
+                _GLOBAL_RESOURCE,
+                expected_label,
+                None,
+                None,
+                None,
+                None,
+            ),
+        )
+
     def test_emit_with_encoded_json(self):
         """
         Handler should parse json encoded as a string
@@ -608,6 +642,62 @@ def test_broken_encoded_dict(self):
         result = _format_and_parse_message(record, handler)
         self.assertEqual(result, message)
 
+    def test_json_fields(self):
+        """
+        record.json_fields should populate the json payload
+        """
+        from google.cloud.logging_v2.handlers.handlers import _format_and_parse_message
+
+        message = "hello"
+        json_fields = {"key": "val"}
+        record = logging.LogRecord("logname", None, None, None, message, None, None)
+        setattr(record, "json_fields", json_fields)
+        handler = logging.StreamHandler()
+        result = _format_and_parse_message(record, handler)
+        self.assertEqual(result, {"message": message, "key": "val"})
+
+    def test_empty_json_fields(self):
+        """
+        empty jsond_field dictionaries should result in a string output
+        """
+        from google.cloud.logging_v2.handlers.handlers import _format_and_parse_message
+
+        message = "hello"
+        record = logging.LogRecord("logname", None, None, None, message, None, None)
+        setattr(record, "json_fields", {})
+        handler = logging.StreamHandler()
+        result = _format_and_parse_message(record, handler)
+        self.assertEqual(result, message)
+
+    def test_json_fields_empty_message(self):
+        """
+        empty message fields should not be added to json_fields dictionaries
+        """
+        from google.cloud.logging_v2.handlers.handlers import _format_and_parse_message
+
+        message = None
+        json_fields = {"key": "val"}
+        record = logging.LogRecord("logname", None, None, None, message, None, None)
+        setattr(record, "json_fields", json_fields)
+        handler = logging.StreamHandler()
+        result = _format_and_parse_message(record, handler)
+        self.assertEqual(result, json_fields)
+
+    def test_json_fields_with_json_message(self):
+        """
+        if json_fields and message are both dicts, they should be combined
+        """
+        from google.cloud.logging_v2.handlers.handlers import _format_and_parse_message
+
+        message = {"key_m": "val_m"}
+        json_fields = {"key_j": "val_j"}
+        record = logging.LogRecord("logname", None, None, None, message, None, None)
+        setattr(record, "json_fields", json_fields)
+        handler = logging.StreamHandler()
+        result = _format_and_parse_message(record, handler)
+        self.assertEqual(result["key_m"], message["key_m"])
+        self.assertEqual(result["key_j"], json_fields["key_j"])
+
 
 class TestSetupLogging(unittest.TestCase):
     def _call_fut(self, handler, excludes=None):
diff --git a/tests/unit/handlers/test_structured_log.py b/tests/unit/handlers/test_structured_log.py
index d9dfa2512..08e4c2906 100644
--- a/tests/unit/handlers/test_structured_log.py
+++ b/tests/unit/handlers/test_structured_log.py
@@ -321,3 +321,26 @@ def test_format_overrides(self):
             result = json.loads(handler.format(record))
             for (key, value) in expected_payload.items():
                 self.assertEqual(value, result[key])
+
+    def test_format_with_json_fields(self):
+        """
+        User can add json_fields to the record, which should populate the payload
+        """
+        import logging
+        import json
+
+        handler = self._make_one()
+        message = "name: %s"
+        name_arg = "Daniel"
+        expected_result = "name: Daniel"
+        json_fields = {"hello": "world", "number": 12}
+        record = logging.LogRecord(
+            None, logging.INFO, None, None, message, name_arg, None,
+        )
+        record.created = None
+        setattr(record, "json_fields", json_fields)
+        handler.filter(record)
+        result = json.loads(handler.format(record))
+        self.assertEqual(result["message"], expected_result)
+        self.assertEqual(result["hello"], "world")
+        self.assertEqual(result["number"], 12)