diff --git a/.gitignore b/.gitignore
index 5125374..8bd6dbd 100644
--- a/.gitignore
+++ b/.gitignore
@@ -27,3 +27,4 @@ openspec/
 .opencode/
 AGENTS.md
 CLAUDE.md
+opencode.json
diff --git a/README.md b/README.md
index 7ef6617..a0a2089 100644
--- a/README.md
+++ b/README.md
@@ -84,11 +84,14 @@ Webconfig supports secure TLS/SSL connections to Kafka brokers for both consumer
 
 **TLS Configuration Options:**
 
-- `tls.enabled` - Enable/disable TLS for Kafka connections (default: false)
-- `tls.cert_file` - Path to client certificate file for mTLS authentication (optional)
-- `tls.key_file` - Path to client private key file for mTLS authentication (optional)
-- `tls.ca_cert_file` - Path to CA certificate file for broker verification (optional)
-- `tls.insecure_skip_verify` - Skip certificate verification (insecure, for testing only, default: false)
+- `tls_enabled` - Enable/disable TLS for Kafka connections (default: false)
+- `tls_cert_file` - Path to client certificate file for mTLS authentication (optional)
+- `tls_key_file` - Path to client private key file for mTLS authentication (optional)
+- `tls_ca_cert_file` - Path to CA certificate file for broker verification (optional)
+- `tls_insecure_skip_verify` - Skip certificate verification (insecure, for testing only, default: false)
+- `tls_server_name` - Optional SNI and certificate name override (default: empty)
+
+Kafka keeps its existing `tls_enabled` flag for each consumer cluster and producer. Cassandra uses its existing `is_ssl_enabled` flag; after TLS is enabled, both integrations use the same flat `tls_*` fields, including identical `tls_insecure_skip_verify` and `tls_server_name` behavior.
 
 **Consumer TLS Configuration Example:**
 
@@ -99,13 +102,12 @@ Webconfig supports secure TLS/SSL connections to Kafka brokers for both consumer
         topics = "config-version-report"
         consumer_group = "webconfig"
 
-        tls {
-            enabled = true
-            cert_file = "/etc/webconfig/kafka/client.crt"
-            key_file = "/etc/webconfig/kafka/client.key"
-            ca_cert_file = "/etc/webconfig/kafka/ca.crt"
-            insecure_skip_verify = false
-        }
+        tls_enabled = true
+        tls_cert_file = "/etc/webconfig/kafka/client.crt"
+        tls_key_file = "/etc/webconfig/kafka/client.key"
+        tls_ca_cert_file = "/etc/webconfig/kafka/ca.crt"
+        tls_insecure_skip_verify = false
+        tls_server_name = "kafka.example.com"
 
         # Per-cluster TLS configuration
         clusters {
@@ -114,12 +116,12 @@ Webconfig supports secure TLS/SSL connections to Kafka brokers for both consumer
                 brokers = "kafka-mesh:9093"
                 topics = "staging-chi-onewifi-from-device"
 
-                tls {
-                    enabled = true
-                    cert_file = "/etc/webconfig/kafka/mesh-client.crt"
-                    key_file = "/etc/webconfig/kafka/mesh-client.key"
-                    ca_cert_file = "/etc/webconfig/kafka/mesh-ca.crt"
-                }
+                tls_enabled = true
+                tls_cert_file = "/etc/webconfig/kafka/mesh-client.crt"
+                tls_key_file = "/etc/webconfig/kafka/mesh-client.key"
+                tls_ca_cert_file = "/etc/webconfig/kafka/mesh-ca.crt"
+                tls_insecure_skip_verify = false
+                tls_server_name = "kafka-mesh.example.com"
             }
         }
     }
@@ -130,23 +132,27 @@ Webconfig supports secure TLS/SSL connections to Kafka brokers for both consumer
 ```shell
     kafka_producer {
         enabled = true
+        // one_line consolidates consumer and producer enqueue success under logger=kafka;
+        // two_line also emits the consumer processing success record and uses
+        // logger=kafkaproducer for producer records.
+        logging_mode = "one_line"
         brokers = "kafka-broker:9093"
         topic = "webconfig_downstream"
 
-        tls {
-            enabled = true
-            cert_file = "/etc/webconfig/kafka/producer-client.crt"
-            key_file = "/etc/webconfig/kafka/producer-client.key"
-            ca_cert_file = "/etc/webconfig/kafka/ca.crt"
-        }
+        tls_enabled = true
+        tls_cert_file = "/etc/webconfig/kafka/producer-client.crt"
+        tls_key_file = "/etc/webconfig/kafka/producer-client.key"
+        tls_ca_cert_file = "/etc/webconfig/kafka/ca.crt"
+        tls_insecure_skip_verify = false
+        tls_server_name = "kafka.example.com"
     }
 ```
 
 **Certificate Requirements:**
 
-1. **Client Certificate (mTLS)**: If `cert_file` and `key_file` are provided, mutual TLS authentication is enabled. The certificate and key must be in PEM format.
+1. **Client Certificate (mTLS)**: If `tls_cert_file` and `tls_key_file` are provided, mutual TLS authentication is enabled. The certificate and key must be in PEM format.
 
-2. **CA Certificate**: If `ca_cert_file` is provided, it will be used to verify the Kafka broker's certificate. This is useful when using self-signed certificates or internal CAs.
+2. **CA Certificate**: If `tls_ca_cert_file` is provided, it will be used to verify the Kafka broker's certificate. This is useful when using self-signed certificates or internal CAs.
 
 3. **Certificate Validation**: All certificate files are validated at startup. The application will fail to start with clear error messages if:
    - Certificate files are missing or unreadable
@@ -156,8 +162,8 @@ Webconfig supports secure TLS/SSL connections to Kafka brokers for both consumer
 **TLS Security Best Practices:**
 
 1. **Use TLS in Production**: Always enable TLS for production Kafka connections to encrypt data in transit
-2. **Use mTLS**: Provide client certificates (`cert_file` and `key_file`) for mutual authentication
-3. **Verify Certificates**: Never use `insecure_skip_verify = true` in production - it disables certificate verification and is insecure
+2. **Use mTLS**: Provide client certificates (`tls_cert_file` and `tls_key_file`) for mutual authentication
+3. **Verify Certificates**: Never use `tls_insecure_skip_verify = true` in production - it disables certificate verification and is insecure
 4. **Protect Certificate Files**: Set appropriate file permissions (0600) on certificate and key files
 5. **Use Secure Ports**: Configure Kafka brokers to listen on secure ports (typically 9093 for TLS)
 6. **Certificate Rotation**: Plan for certificate rotation - the service must be restarted to pick up new certificates
diff --git a/common/kafka_tls.go b/common/kafka_tls.go
index bcc182a..9c76a31 100644
--- a/common/kafka_tls.go
+++ b/common/kafka_tls.go
@@ -27,109 +27,115 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
-// LoadKafkaTLSConfig loads TLS configuration from HOCON config at the specified prefix.
-// Prefix should be like "webconfig.kafka" or "webconfig.kafka.clusters.mesh" or "webconfig.kafka_producer".
-// Returns nil if TLS is not enabled.
-// Returns error if TLS is enabled but configuration is invalid.
-func LoadKafkaTLSConfig(conf *configuration.Config, prefix string) (*tls.Config, error) {
-	tlsEnabled := conf.GetBoolean(prefix + ".tls_enabled")
-	if !tlsEnabled {
-		return nil, nil
+// LoadTLSConfig loads the shared flat tls_* configuration used by Kafka and Cassandra.
+// Prefix should be empty for a driver sub-config, or a full HOCON path such as
+// "webconfig.kafka" or "webconfig.kafka_producer".
+func LoadTLSConfig(conf *configuration.Config, prefix, component string) (*tls.Config, error) {
+	key := func(name string) string {
+		if prefix == "" {
+			return name
+		}
+		return prefix + "." + name
 	}
 
-	tlsConfig := &tls.Config{}
-
-	// Check insecure_skip_verify flag first
-	insecureSkipVerify := conf.GetBoolean(prefix + ".tls_insecure_skip_verify")
+	insecureSkipVerify := conf.GetBoolean(key("tls_insecure_skip_verify"))
+	certFile := conf.GetString(key("tls_cert_file"))
+	keyFile := conf.GetString(key("tls_key_file"))
+	caCertFile := conf.GetString(key("tls_ca_cert_file"))
+	serverName := conf.GetString(key("tls_server_name"))
+
+	cipherSuites := []uint16{
+		tls.TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384,
+		tls.TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384,
+		tls.TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256,
+		tls.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,
+		tls.TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305_SHA256,
+		tls.TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256,
+	}
+	if component == "Cassandra" {
+		cipherSuites = append(cipherSuites, tls.TLS_RSA_WITH_AES_128_CBC_SHA)
+	}
 
-	// Load client certificates for mTLS if provided (optional when insecure_skip_verify is true)
-	certFile := conf.GetString(prefix + ".tls_cert_file")
-	keyFile := conf.GetString(prefix + ".tls_key_file")
+	tlsConfig := &tls.Config{
+		MinVersion:         tls.VersionTLS12,
+		ServerName:         serverName,
+		CipherSuites:       cipherSuites,
+		InsecureSkipVerify: insecureSkipVerify,
+	}
 
-	// When insecure_skip_verify is true and no cert files configured, skip loading certificates
-	// This allows TLS without client authentication (server-only TLS)
 	if insecureSkipVerify && (len(certFile) == 0 || len(keyFile) == 0) {
-		// Insecure mode without client certificates - skip cert loading
+		log.WithField("component", component).Warn("TLS enabled in insecure mode without client certificates")
 	} else if len(certFile) > 0 && len(keyFile) > 0 {
-		// Only validate cert files exist when verification is enabled
 		if !insecureSkipVerify {
-			// Validate certificate file exists
 			if _, err := os.Stat(certFile); os.IsNotExist(err) {
 				return nil, NewError(fmt.Errorf("TLS certificate file does not exist: %s", certFile))
 			}
-
-			// Validate key file exists
 			if _, err := os.Stat(keyFile); os.IsNotExist(err) {
 				return nil, NewError(fmt.Errorf("TLS key file does not exist: %s", keyFile))
 			}
 		}
 
-		// Load and parse the certificate and key
 		cert, err := tls.LoadX509KeyPair(certFile, keyFile)
 		if err != nil {
 			return nil, NewError(fmt.Errorf("failed to load TLS certificate and key from %s and %s: %v", certFile, keyFile, err))
 		}
-
 		tlsConfig.Certificates = []tls.Certificate{cert}
 		log.WithFields(log.Fields{
-			"prefix":    prefix,
+			"component": component,
 			"cert_file": certFile,
 			"key_file":  keyFile,
 		}).Info("Loaded TLS client certificate for mTLS")
 	} else if len(certFile) > 0 || len(keyFile) > 0 {
-		// Partial cert configuration detected - require both cert and key when verification is enabled
 		if !insecureSkipVerify {
 			return nil, NewError(fmt.Errorf("TLS enabled with verification but incomplete certificate configuration (cert: %s, key: %s)", certFile, keyFile))
 		}
 	}
 
-	// Load CA certificate if provided (optional when insecure_skip_verify is true)
-	caCertFile := conf.GetString(prefix + ".tls_ca_cert_file")
-	// When insecure_skip_verify is true and no CA file configured, skip loading CA cert
-	// This allows TLS without broker verification (insecure mode)
 	if insecureSkipVerify && len(caCertFile) == 0 {
-		// Insecure mode without CA cert - skip CA loading
+		log.WithField("component", component).Warn("TLS enabled in insecure mode without CA certificate")
 	} else if len(caCertFile) > 0 {
-		// Only validate CA cert file exists when verification is enabled
 		if !insecureSkipVerify {
-			// Validate CA certificate file exists
 			if _, err := os.Stat(caCertFile); os.IsNotExist(err) {
 				return nil, NewError(fmt.Errorf("TLS CA certificate file does not exist: %s", caCertFile))
 			}
 		}
 
-		// Load CA certificate
 		caCert, err := os.ReadFile(caCertFile)
 		if err != nil {
 			return nil, NewError(fmt.Errorf("failed to read TLS CA certificate from %s: %v", caCertFile, err))
 		}
-
-		// Parse CA certificate
 		caCertPool := x509.NewCertPool()
 		if !caCertPool.AppendCertsFromPEM(caCert) {
 			return nil, NewError(fmt.Errorf("failed to parse TLS CA certificate from %s", caCertFile))
 		}
-
 		tlsConfig.RootCAs = caCertPool
 		log.WithFields(log.Fields{
-			"prefix":       prefix,
+			"component":    component,
 			"ca_cert_file": caCertFile,
-		}).Info("Loaded TLS CA certificate for broker verification")
+		}).Info("Loaded TLS CA certificate for server verification")
 	}
 
 	if insecureSkipVerify {
-		tlsConfig.InsecureSkipVerify = true
-		log.WithFields(log.Fields{
-			"prefix": prefix,
-		}).Warn("TLS certificate verification is disabled (insecure_skip_verify=true). This is insecure and should only be used for testing.")
+		log.WithField("component", component).Warn("TLS certificate verification is disabled (insecure_skip_verify=true). This is insecure and should only be used for testing.")
 	}
-
 	log.WithFields(log.Fields{
-		"prefix":               prefix,
+		"component":            component,
 		"has_client_cert":      len(tlsConfig.Certificates) > 0,
 		"has_ca_cert":          tlsConfig.RootCAs != nil,
 		"insecure_skip_verify": insecureSkipVerify,
-	}).Info("TLS configuration loaded for Kafka connection")
+		"server_name":          serverName,
+	}).Info("TLS configuration loaded")
 
 	return tlsConfig, nil
 }
+
+// LoadKafkaTLSConfig loads enabled Kafka TLS configuration from HOCON.
+// Prefix should be like "webconfig.kafka", "webconfig.kafka.clusters.mesh",
+// or "webconfig.kafka_producer".
+func LoadKafkaTLSConfig(conf *configuration.Config, prefix string) (*tls.Config, error) {
+	tlsEnabled := conf.GetBoolean(prefix + ".tls_enabled")
+	if !tlsEnabled {
+		return nil, nil
+	}
+	return LoadTLSConfig(conf, prefix, "Kafka")
+}
diff --git a/common/kafka_tls_test.go b/common/kafka_tls_test.go
index 688d07d..46cd2bb 100644
--- a/common/kafka_tls_test.go
+++ b/common/kafka_tls_test.go
@@ -20,6 +20,7 @@ package common
 import (
 	"crypto/rand"
 	"crypto/rsa"
+	"crypto/tls"
 	"crypto/x509"
 	"crypto/x509/pkix"
 	"encoding/pem"
@@ -325,6 +326,7 @@ webconfig {
 		tls_cert_file = "` + certFile + `"
 		tls_key_file = "` + keyFile + `"
 		tls_ca_cert_file = "` + caCertFile + `"
+			tls_server_name = "kafka.example.com"
 	}
 }
 `
@@ -336,6 +338,38 @@ webconfig {
 	assert.Assert(t, len(tlsConfig.Certificates) == 1, "Should have one client certificate")
 	assert.Assert(t, tlsConfig.RootCAs != nil, "Should have custom CA")
 	assert.Assert(t, !tlsConfig.InsecureSkipVerify, "Should not skip verification")
+	assert.Equal(t, tlsConfig.ServerName, "kafka.example.com")
+	assert.Equal(t, tlsConfig.MinVersion, uint16(0x0303))
+	assert.Assert(t, len(tlsConfig.CipherSuites) > 0, "Should configure TLS cipher suites")
+	assert.Assert(t, !containsCipherSuite(tlsConfig.CipherSuites, tls.TLS_RSA_WITH_AES_128_CBC_SHA), "Kafka should not enable the Cassandra legacy cipher suite")
+}
+
+func TestLoadTLSConfig_CassandraIncludesLegacyCipher(t *testing.T) {
+	conf := configuration.ParseString(`
+webconfig {
+	database {
+		cassandra {
+			tls_server_name = "cassandra.example.com"
+			tls_insecure_skip_verify = false
+		}
+	}
+}
+`)
+
+	tlsConfig, err := LoadTLSConfig(conf.GetConfig("webconfig.database.cassandra"), "", "Cassandra")
+
+	assert.NilError(t, err)
+	assert.Equal(t, tlsConfig.ServerName, "cassandra.example.com")
+	assert.Assert(t, containsCipherSuite(tlsConfig.CipherSuites, tls.TLS_RSA_WITH_AES_128_CBC_SHA), "Cassandra should retain its legacy 3.11.x cipher suite")
+}
+
+func containsCipherSuite(cipherSuites []uint16, wanted uint16) bool {
+	for _, cipherSuite := range cipherSuites {
+		if cipherSuite == wanted {
+			return true
+		}
+	}
+	return false
 }
 
 func TestLoadKafkaTLSConfig_DifferentPrefixes(t *testing.T) {
diff --git a/config/sample_webconfig.conf b/config/sample_webconfig.conf
index 04eebcb..48fcfeb 100644
--- a/config/sample_webconfig.conf
+++ b/config/sample_webconfig.conf
@@ -162,6 +162,12 @@ webconfig {
             enabled = false
         }
 
+        // Protect /api/v1/token with API token authentication.
+        // Defaults to false to preserve existing endpoint access.
+        token_api_token_auth {
+            enabled = false
+        }
+
         device_api_token_auth {
             enabled = true
         }
@@ -286,14 +292,16 @@ webconfig {
 
         // TLS configuration for secure Kafka connections
         // NOTE: These are flat keys read by LoadKafkaTLSConfig at prefix "webconfig.kafka"
-        // Certificate files (tls_cert_file, tls_key_file, tls_ca_cert_file) are REQUIRED when
-        // tls_insecure_skip_verify=false (default). They are OPTIONAL when tls_insecure_skip_verify=true.
+        // Client certificate/key are optional unless the Kafka broker requires mTLS.
+        // tls_ca_cert_file is optional; when unset, the system CA pool is used.
         tls_enabled = false
         tls_cert_file = "/etc/webconfig/kafka/client-cert.pem"
         tls_key_file = "/etc/webconfig/kafka/client-key.pem"
         tls_ca_cert_file = "/etc/webconfig/kafka/ca-cert.pem"
         // WARNING: set true only in dev/test environments, NEVER in production
         tls_insecure_skip_verify = false
+        // Optional SNI/server certificate name. Leave empty to use the broker address.
+        tls_server_name = ""
 
         // Reconnect resilience — tuned for broker restart recovery.
 
@@ -359,6 +367,7 @@ webconfig {
                 tls_key_file = "/etc/webconfig/kafka/mesh-client-key.pem"
                 tls_ca_cert_file = "/etc/webconfig/kafka/mesh-ca-cert.pem"
                 tls_insecure_skip_verify = false
+                tls_server_name = ""
             }
             east {
                 enabled = false
@@ -379,6 +388,7 @@ webconfig {
                 tls_key_file = "/etc/webconfig/kafka/east-client-key.pem"
                 tls_ca_cert_file = "/etc/webconfig/kafka/east-ca-cert.pem"
                 tls_insecure_skip_verify = false
+                tls_server_name = ""
             }
         }
     }
@@ -406,6 +416,10 @@ webconfig {
     // forward kafka messages if needed
     kafka_producer {
         enabled = false
+        // Kafka success logging: "one_line" consolidates consumer and producer enqueue success
+        // under logger=kafka; "two_line" also emits the consumer processing success record and
+        // uses logger=kafkaproducer for producer records.
+        logging_mode = "one_line"
         brokers = "localhost:9092"
         topic = "webconfig_downstream"
 
@@ -415,6 +429,7 @@ webconfig {
         tls_key_file = "/etc/webconfig/kafka/producer-client-key.pem"
         tls_ca_cert_file = "/etc/webconfig/kafka/producer-ca-cert.pem"
         tls_insecure_skip_verify = false
+        tls_server_name = ""
 
         // Reconnect resilience — tuned for broker restart recovery.
 
diff --git a/db/cassandra/cassandra_client.go b/db/cassandra/cassandra_client.go
index 6c7f6d7..a78bdbd 100644
--- a/db/cassandra/cassandra_client.go
+++ b/db/cassandra/cassandra_client.go
@@ -19,8 +19,6 @@ package cassandra
 
 import (
 	"context"
-	"crypto/tls"
-	"crypto/x509"
 	"errors"
 	"fmt"
 	"os"
@@ -33,7 +31,6 @@ import (
 	"github.com/rdkcentral/webconfig/db"
 	"github.com/rdkcentral/webconfig/security"
 	"github.com/rdkcentral/webconfig/util"
-	log "github.com/sirupsen/logrus"
 )
 
 const (
@@ -165,7 +162,7 @@ func NewCassandraClient(conf *configuration.Config, testOnly bool) (*CassandraCl
 
 	if isSslEnabled {
 		insecureSkipVerify := dbconf.GetBoolean("tls_insecure_skip_verify")
-		tlsConfig, err := loadCassandraTLSConfig(dbconf, dbdriver, insecureSkipVerify)
+		tlsConfig, err := common.LoadTLSConfig(dbconf, "", "Cassandra")
 		if err != nil {
 			return nil, common.NewError(err)
 		}
@@ -214,128 +211,6 @@ func NewCassandraClient(conf *configuration.Config, testOnly bool) (*CassandraCl
 	}, nil
 }
 
-// loadCassandraTLSConfig loads TLS configuration for Cassandra connection.
-// Returns a tls.Config with certificates loaded from the configuration.
-// Reads flat tls_* keys directly under the database driver config (cassandra or yugabyte).
-func loadCassandraTLSConfig(dbconf *configuration.Config, dbdriver string, insecureSkipVerify bool) (*tls.Config, error) {
-	// Load client certificates for mTLS if provided (optional when insecure_skip_verify is true)
-	certFile := dbconf.GetString("tls_cert_file")
-	keyFile := dbconf.GetString("tls_key_file")
-	caCertFile := dbconf.GetString("tls_ca_cert_file")
-	serverName := dbconf.GetString("tls_server_name")
-
-	// Create TLS config for Cassandra connection.
-	// Prefer modern ECDHE+AEAD suites for forward secrecy; keep TLS_RSA_WITH_AES_128_CBC_SHA
-	// last as a fallback for legacy Cassandra 3.11.x nodes that only negotiate that suite.
-	tlsConfig := &tls.Config{
-		MinVersion: tls.VersionTLS12,
-		CipherSuites: []uint16{
-			tls.TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384,
-			tls.TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384,
-			tls.TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256,
-			tls.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,
-			tls.TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305_SHA256,
-			tls.TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256,
-			// Legacy fallback for Cassandra 3.11.x — do not remove
-			tls.TLS_RSA_WITH_AES_128_CBC_SHA,
-		},
-		InsecureSkipVerify: insecureSkipVerify,
-		ServerName:         serverName,
-	}
-
-	// When insecure_skip_verify is true and no cert files configured, skip loading certificates
-	// This allows TLS without client authentication (server-only TLS)
-	if insecureSkipVerify && (len(certFile) == 0 || len(keyFile) == 0) {
-		// Insecure mode without client certificates - skip cert loading
-		log.WithFields(log.Fields{
-			"driver": dbdriver,
-		}).Warn("Cassandra TLS enabled in insecure mode without client certificates")
-	} else if len(certFile) > 0 && len(keyFile) > 0 {
-		// Only validate cert files exist when verification is enabled
-		if !insecureSkipVerify {
-			// Validate certificate file exists
-			if _, err := os.Stat(certFile); os.IsNotExist(err) {
-				return nil, fmt.Errorf("Cassandra TLS certificate file does not exist: %s", certFile)
-			}
-
-			// Validate key file exists
-			if _, err := os.Stat(keyFile); os.IsNotExist(err) {
-				return nil, fmt.Errorf("Cassandra TLS key file does not exist: %s", keyFile)
-			}
-		}
-
-		// Load and parse the certificate and key
-		cert, err := tls.LoadX509KeyPair(certFile, keyFile)
-		if err != nil {
-			return nil, fmt.Errorf("failed to load Cassandra TLS certificate and key from %s and %s: %v", certFile, keyFile, err)
-		}
-
-		tlsConfig.Certificates = []tls.Certificate{cert}
-		log.WithFields(log.Fields{
-			"driver":    dbdriver,
-			"cert_file": certFile,
-			"key_file":  keyFile,
-		}).Info("Loaded Cassandra TLS client certificate for mTLS")
-	} else if len(certFile) > 0 || len(keyFile) > 0 {
-		// Partial cert configuration detected - require both cert and key when verification is enabled
-		if !insecureSkipVerify {
-			return nil, fmt.Errorf("Cassandra TLS enabled with verification but incomplete certificate configuration (cert: %s, key: %s)", certFile, keyFile)
-		}
-	}
-
-	// Load CA certificate if provided (optional when insecure_skip_verify is true)
-	// When insecure_skip_verify is true and no CA file configured, skip loading CA cert
-	// This allows TLS without server verification (insecure mode)
-	if insecureSkipVerify && len(caCertFile) == 0 {
-		// Insecure mode without CA cert - skip CA loading
-		log.WithFields(log.Fields{
-			"driver": dbdriver,
-		}).Warn("Cassandra TLS enabled in insecure mode without CA certificate")
-	} else if len(caCertFile) > 0 {
-		// Only validate CA cert file exists when verification is enabled
-		if !insecureSkipVerify {
-			// Validate CA certificate file exists
-			if _, err := os.Stat(caCertFile); os.IsNotExist(err) {
-				return nil, fmt.Errorf("Cassandra TLS CA certificate file does not exist: %s", caCertFile)
-			}
-		}
-
-		// Load CA certificate
-		caCert, err := os.ReadFile(caCertFile)
-		if err != nil {
-			return nil, fmt.Errorf("failed to read Cassandra TLS CA certificate from %s: %v", caCertFile, err)
-		}
-
-		// Parse CA certificate
-		caCertPool := x509.NewCertPool()
-		if !caCertPool.AppendCertsFromPEM(caCert) {
-			return nil, fmt.Errorf("failed to parse Cassandra TLS CA certificate from %s", caCertFile)
-		}
-
-		tlsConfig.RootCAs = caCertPool
-		log.WithFields(log.Fields{
-			"driver":       dbdriver,
-			"ca_cert_file": caCertFile,
-		}).Info("Loaded Cassandra TLS CA certificate for server verification")
-	}
-
-	if insecureSkipVerify {
-		log.WithFields(log.Fields{
-			"driver": dbdriver,
-		}).Warn("Cassandra TLS certificate verification is disabled (insecure_skip_verify=true). This is insecure and should only be used for testing.")
-	}
-
-	log.WithFields(log.Fields{
-		"driver":               dbdriver,
-		"has_client_cert":      len(tlsConfig.Certificates) > 0,
-		"has_ca_cert":          tlsConfig.RootCAs != nil,
-		"insecure_skip_verify": insecureSkipVerify,
-		"cipher_suites":        len(tlsConfig.CipherSuites),
-	}).Info("TLS configuration loaded for Cassandra connection")
-
-	return tlsConfig, nil
-}
-
 func (c *CassandraClient) Codec() *security.AesCodec {
 	return c.AesCodec
 }
diff --git a/go.mod b/go.mod
index 1a7ca88..47b6ec3 100644
--- a/go.mod
+++ b/go.mod
@@ -16,14 +16,14 @@ require (
 	github.com/twmb/murmur3 v1.1.8
 	github.com/vmihailenco/msgpack v4.0.4+incompatible
 	github.com/vmihailenco/msgpack/v4 v4.3.12
-	go.opentelemetry.io/otel v1.43.0
+	go.opentelemetry.io/otel v1.44.0
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.43.0
 	go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.43.0
-	go.opentelemetry.io/otel/sdk v1.43.0
-	go.opentelemetry.io/otel/trace v1.43.0
+	go.opentelemetry.io/otel/sdk v1.44.0
+	go.opentelemetry.io/otel/trace v1.44.0
 	go.uber.org/automaxprocs v1.6.0
 	go.uber.org/ratelimit v0.3.1
-	golang.org/x/sync v0.20.0
+	golang.org/x/sync v0.22.0
 	gotest.tools v2.2.0+incompatible
 	modernc.org/sqlite v1.50.0
 )
@@ -63,17 +63,17 @@ require (
 	github.com/vmihailenco/tagparser v0.1.2 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.43.0 // indirect
-	go.opentelemetry.io/otel/metric v1.43.0 // indirect
+	go.opentelemetry.io/otel/metric v1.44.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.10.0 // indirect
 	go.yaml.in/yaml/v2 v2.4.4 // indirect
-	golang.org/x/crypto v0.52.0 // indirect
-	golang.org/x/net v0.55.0 // indirect
-	golang.org/x/sys v0.45.0 // indirect
-	golang.org/x/text v0.37.0 // indirect
+	golang.org/x/crypto v0.55.0 // indirect
+	golang.org/x/net v0.58.0 // indirect
+	golang.org/x/sys v0.47.0 // indirect
+	golang.org/x/text v0.41.0 // indirect
 	google.golang.org/appengine v1.6.8 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20260504160031-60b97b32f348 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20260504160031-60b97b32f348 // indirect
-	google.golang.org/grpc v1.82.1 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect
+	google.golang.org/grpc v1.83.2 // indirect
 	google.golang.org/protobuf v1.36.11 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	modernc.org/libc v1.72.2 // indirect
diff --git a/go.sum b/go.sum
index 6af8102..5d4c236 100644
--- a/go.sum
+++ b/go.sum
@@ -138,22 +138,22 @@ github.com/vmihailenco/tagparser v0.1.2/go.mod h1:OeAg3pn3UbLjkWt+rN9oFYB6u/cQgq
 github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
 go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64=
 go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y=
-go.opentelemetry.io/otel v1.43.0 h1:mYIM03dnh5zfN7HautFE4ieIig9amkNANT+xcVxAj9I=
-go.opentelemetry.io/otel v1.43.0/go.mod h1:JuG+u74mvjvcm8vj8pI5XiHy1zDeoCS2LB1spIq7Ay0=
+go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU=
+go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc=
 go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.43.0 h1:88Y4s2C8oTui1LGM6bTWkw0ICGcOLCAI5l6zsD1j20k=
 go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.43.0/go.mod h1:Vl1/iaggsuRlrHf/hfPJPvVag77kKyvrLeD10kpMl+A=
 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.43.0 h1:3iZJKlCZufyRzPzlQhUIWVmfltrXuGyfjREgGP3UUjc=
 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.43.0/go.mod h1:/G+nUPfhq2e+qiXMGxMwumDrP5jtzU+mWN7/sjT2rak=
 go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.43.0 h1:mS47AX77OtFfKG4vtp+84kuGSFZHTyxtXIN269vChY0=
 go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.43.0/go.mod h1:PJnsC41lAGncJlPUniSwM81gc80GkgWJWr3cu2nKEtU=
-go.opentelemetry.io/otel/metric v1.43.0 h1:d7638QeInOnuwOONPp4JAOGfbCEpYb+K6DVWvdxGzgM=
-go.opentelemetry.io/otel/metric v1.43.0/go.mod h1:RDnPtIxvqlgO8GRW18W6Z/4P462ldprJtfxHxyKd2PY=
-go.opentelemetry.io/otel/sdk v1.43.0 h1:pi5mE86i5rTeLXqoF/hhiBtUNcrAGHLKQdhg4h4V9Dg=
-go.opentelemetry.io/otel/sdk v1.43.0/go.mod h1:P+IkVU3iWukmiit/Yf9AWvpyRDlUeBaRg6Y+C58QHzg=
-go.opentelemetry.io/otel/sdk/metric v1.43.0 h1:S88dyqXjJkuBNLeMcVPRFXpRw2fuwdvfCGLEo89fDkw=
-go.opentelemetry.io/otel/sdk/metric v1.43.0/go.mod h1:C/RJtwSEJ5hzTiUz5pXF1kILHStzb9zFlIEe85bhj6A=
-go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09nk+3A=
-go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0=
+go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc=
+go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo=
+go.opentelemetry.io/otel/sdk v1.44.0 h1:nHYwb9lK+fJPU/dnT6s7W7Z8itMWyqrnVfbheVYrZ58=
+go.opentelemetry.io/otel/sdk v1.44.0/go.mod h1:Osuydd3Se74nqjAKxid74N5eC+jfEqfTegHRnq58oK0=
+go.opentelemetry.io/otel/sdk/metric v1.44.0 h1:3LlKgI+VjbVsjNRFZJZAJ30WjXC5VkNRks6si09iEfI=
+go.opentelemetry.io/otel/sdk/metric v1.44.0/go.mod h1:5B5pMARnXxKhltooO4xUuCBorl65a4EpnTalObqOigA=
+go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk=
+go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE=
 go.opentelemetry.io/proto/otlp v1.10.0 h1:IQRWgT5srOCYfiWnpqUYz9CVmbO8bFmKcwYxpuCSL2g=
 go.opentelemetry.io/proto/otlp v1.10.0/go.mod h1:/CV4QoCR/S9yaPj8utp3lvQPoqMtxXdzn7ozvvozVqk=
 go.uber.org/atomic v1.7.0 h1:ADUqmZGgLDDfbSL9ZmPxKTybcoEYHgpYfELNoN+7hsw=
@@ -169,11 +169,11 @@ go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.6.0/go.mod h1:OFC/31mSvZgRz0V1QTNCzfAI1aIRzbiufJtkMIlEp58=
-golang.org/x/crypto v0.52.0 h1:RMs7fP2rXdep0CftQlK8Uf+kibLm7qkCcradZWYz988=
-golang.org/x/crypto v0.52.0/go.mod h1:1QgfPxDqh0T2M/elOJtp9RvuR95kVjir0e6/BvEmGbc=
+golang.org/x/crypto v0.55.0 h1:+KWHjbgOaAQ66dh/YlkZKHlz9ZUlq61AFirAR9ntP8M=
+golang.org/x/crypto v0.55.0/go.mod h1:uq0V9dE/fzQuJtbnL+2EhWOE63vo164FY8xqEnV9xis=
 golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
-golang.org/x/mod v0.35.0 h1:Ww1D637e6Pg+Zb2KrWfHQUnH2dQRLBQyAtpr/haaJeM=
-golang.org/x/mod v0.35.0/go.mod h1:+GwiRhIInF8wPm+4AoT6L0FA1QWAad3OMdTRx4tFYlU=
+golang.org/x/mod v0.38.0 h1:MECBjubtXD7yj4HrhIUcywNaGeNVUdfVnxmPajOk4yk=
+golang.org/x/mod v0.38.0/go.mod h1:V6Xz0pq8TQ3dGqVQ1FVHuelZpAL0uNhSkk9ogYP3c40=
 golang.org/x/net v0.0.0-20190603091049-60506f45cf65/go.mod h1:HSz+uSET+XFnRR8LxR5pz3Of3rY3CfYBVs4xY44aLks=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200114155413-6afb5195e5aa/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
@@ -182,20 +182,20 @@ golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v
 golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
-golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8=
-golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww=
+golang.org/x/net v0.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To=
+golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
-golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
+golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek=
+golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY=
-golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
+golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
+golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
@@ -205,13 +205,13 @@ golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
 golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
-golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc=
-golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38=
+golang.org/x/text v0.41.0 h1:vz/seA0lnX87Othu2f/0L24RcgrXD9/YFTSuGjj3rH8=
+golang.org/x/text v0.41.0/go.mod h1:jvf1O8ajNzZqhSrQBPbutR/EB83Cc0CFrezNQIwbb5M=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
-golang.org/x/tools v0.44.0 h1:UP4ajHPIcuMjT1GqzDWRlalUEoY+uzoZKnhOjbIPD2c=
-golang.org/x/tools v0.44.0/go.mod h1:KA0AfVErSdxRZIsOVipbv3rQhVXTnlU6UhKxHd1seDI=
+golang.org/x/tools v0.48.0 h1:3+hClM1aLL5mjMKm5ovokw9epgRXPuu2tILgismM6RE=
+golang.org/x/tools v0.48.0/go.mod h1:08xX0orndb/F7jJxGDicx061tyd5pcMto75YMAXr6lk=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4=
@@ -219,12 +219,12 @@ gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E
 google.golang.org/appengine v1.6.5/go.mod h1:8WjMMxjGQR8xUklV/ARdw2HLXBOI7O7uCIDZVag1xfc=
 google.golang.org/appengine v1.6.8 h1:IhEN5q69dyKagZPYMSdIjS2HqprW324FRQZJcGqPAsM=
 google.golang.org/appengine v1.6.8/go.mod h1:1jJ3jBArFh5pcgW8gCtRJnepW8FzD1V44FJffLiz/Ds=
-google.golang.org/genproto/googleapis/api v0.0.0-20260504160031-60b97b32f348 h1:U8orV30l6KpDsi9dxU0CoJZGbjS8EEpw+6ba+XwGPQA=
-google.golang.org/genproto/googleapis/api v0.0.0-20260504160031-60b97b32f348/go.mod h1:Yzdzr5OOZFgSsEV2D/Xi9NL3bszpXFAg0hFJiRohcD8=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20260504160031-60b97b32f348 h1:pfIbyB44sWzHiCpRqIen67ZQnVXSfIxWrqUMk1qwODE=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20260504160031-60b97b32f348/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8=
-google.golang.org/grpc v1.82.1 h1:NnAxzGRA0677vCa4BUkOAnO5+FfQqVl9iUXeD0IqcGE=
-google.golang.org/grpc v1.82.1/go.mod h1:yzTZ1TB1Z3SG+LIYaI+WiE8D5+PZ3ArnrSp8zF3+/ZA=
+google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa h1:Kjn0N0tCrDgiAFW+lGO4JZ3ck44CehvJQMAwj9QF0G8=
+google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:q4lMZS6kskjT5HvCPrnnypcDPVJqT/f4nfxmkE7gryY=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa h1:mZHHdPZl0dbGHCflZgAq/Q468DWVFcU2whhB2KAo8fk=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8=
+google.golang.org/grpc v1.83.2 h1:EManeRomTObA0BU7I8vXgg/78uE5MJ9M8B39EX2WscU=
+google.golang.org/grpc v1.83.2/go.mod h1:YPI1hK3kDked6iHvgX3tR0y+nX/qpMFKhPgFsokw1S8=
 google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
 google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
 google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
diff --git a/http/router.go b/http/router.go
index d3e621c..ac8f7f9 100644
--- a/http/router.go
+++ b/http/router.go
@@ -32,14 +32,24 @@ func (s *WebconfigServer) AddBaseRoutes(testOnly bool, router *mux.Router) {
 	r2.HandleFunc("", s.VersionHandler).Methods("GET")
 
 	r3 := router.Path("/config").Subrouter()
-	if s.ConfigApiTokenAuthEnabled() {
+	if testOnly {
+		r3.Use(s.TestingMiddleware)
+	} else if s.ConfigApiTokenAuthEnabled() {
 		r3.Use(s.ApiMiddleware)
+	} else {
+		r3.Use(s.NoAuthMiddleware)
 	}
 	r3.HandleFunc("", s.ServerConfigHandler).Methods("GET")
 
 	if s.TokenApiEnabled() {
 		r4 := router.Path("/api/v1/token").Subrouter()
-		r4.Use(s.NoAuthMiddleware)
+		if testOnly {
+			r4.Use(s.TestingMiddleware)
+		} else if s.TokenApiTokenAuthEnabled() {
+			r4.Use(s.ApiMiddleware)
+		} else {
+			r4.Use(s.NoAuthMiddleware)
+		}
 		r4.HandleFunc("", s.CreateTokenHandler).Methods("POST")
 	}
 
diff --git a/http/webconfig_server.go b/http/webconfig_server.go
index 5277cd1..2755954 100644
--- a/http/webconfig_server.go
+++ b/http/webconfig_server.go
@@ -65,6 +65,7 @@ const (
 	FactoryResetEnabledDefault           = false
 	serverApiTokenAuthEnabledDefault     = true
 	configApiTokenAuthEnabledDefault     = false
+	tokenApiTokenAuthEnabledDefault      = false
 	deviceApiTokenAuthEnabledDefault     = true
 	tokenApiEnabledDefault               = false
 	activeDriverDefault                  = "cassandra"
@@ -74,6 +75,10 @@ const (
 	defaultSupplementaryAppendingEnabled = true
 	authPrefixLength                     = 60
 	defaultMaxRequestBodyBytes           = 1048576
+	maxKafkaProducerLogPayloadBytes      = 4096
+	KafkaLoggingModeOneLine              = "one_line"
+	KafkaLoggingModeTwoLine              = "two_line"
+	defaultKafkaLoggingMode              = KafkaLoggingModeOneLine
 )
 
 var (
@@ -111,6 +116,7 @@ type WebconfigServer struct {
 	factoryResetEnabled           bool
 	serverApiTokenAuthEnabled     bool
 	configApiTokenAuthEnabled     bool
+	tokenApiTokenAuthEnabled      bool
 	deviceApiTokenAuthEnabled     bool
 	tokenApiEnabled               bool
 	kafkaEnabled                  bool
@@ -124,6 +130,7 @@ type WebconfigServer struct {
 	supplementaryAppendingEnabled bool
 	kafkaProducerEnabled          bool
 	kafkaProducerTopic            string
+	kafkaLoggingMode              string
 	upstreamProfilesEnabled       bool
 	queryParamsValidationEnabled  bool
 	minTrust                      int
@@ -273,6 +280,7 @@ func NewWebconfigServer(sc *common.ServerConfig, testOnly bool) *WebconfigServer
 
 	serverApiTokenAuthEnabled := conf.GetBoolean("webconfig.jwt.server_api_token_auth.enabled", serverApiTokenAuthEnabledDefault)
 	configApiTokenAuthEnabled := conf.GetBoolean("webconfig.jwt.config_api_token_auth.enabled", configApiTokenAuthEnabledDefault)
+	tokenApiTokenAuthEnabled := conf.GetBoolean("webconfig.jwt.token_api_token_auth.enabled", tokenApiTokenAuthEnabledDefault)
 	if conf.GetNode("webconfig.jwt.server_api_token_auth.enabled") == nil {
 		log.Warn("webconfig.jwt.server_api_token_auth.enabled is not set in config; defaulting to true (server API token auth enforced). See MIGRATION.md.")
 	}
@@ -306,6 +314,13 @@ func NewWebconfigServer(sc *common.ServerConfig, testOnly bool) *WebconfigServer
 	var kafkaProducer sarama.AsyncProducer
 	kafkaProducerEnabled := conf.GetBoolean("webconfig.kafka_producer.enabled")
 	var kafkaProducerTopic string
+	kafkaLoggingMode := conf.GetString("webconfig.kafka_producer.logging_mode", defaultKafkaLoggingMode)
+	if kafkaLoggingMode != KafkaLoggingModeOneLine && kafkaLoggingMode != KafkaLoggingModeTwoLine {
+		if kafkaProducerEnabled {
+			panic(fmt.Errorf("webconfig.kafka_producer.logging_mode must be %q or %q", KafkaLoggingModeOneLine, KafkaLoggingModeTwoLine))
+		}
+		kafkaLoggingMode = defaultKafkaLoggingMode
+	}
 	if kafkaProducerEnabled {
 		brokersStr := conf.GetString("webconfig.kafka_producer.brokers")
 		if len(brokersStr) == 0 {
@@ -403,6 +418,7 @@ func NewWebconfigServer(sc *common.ServerConfig, testOnly bool) *WebconfigServer
 		factoryResetEnabled:           factoryResetEnabled,
 		serverApiTokenAuthEnabled:     serverApiTokenAuthEnabled,
 		configApiTokenAuthEnabled:     configApiTokenAuthEnabled,
+		tokenApiTokenAuthEnabled:      tokenApiTokenAuthEnabled,
 		deviceApiTokenAuthEnabled:     deviceApiTokenAuthEnabled,
 		tokenApiEnabled:               tokenApiEnabled,
 		kafkaEnabled:                  kafkaEnabled,
@@ -414,6 +430,7 @@ func NewWebconfigServer(sc *common.ServerConfig, testOnly bool) *WebconfigServer
 		supplementaryAppendingEnabled: supplementaryAppendingEnabled,
 		kafkaProducerEnabled:          kafkaProducerEnabled,
 		kafkaProducerTopic:            kafkaProducerTopic,
+		kafkaLoggingMode:              kafkaLoggingMode,
 		upstreamProfilesEnabled:       upstreamProfilesEnabled,
 		queryParamsValidationEnabled:  queryParamsValidationEnabled,
 		minTrust:                      minTrust,
@@ -677,6 +694,14 @@ func (s *WebconfigServer) SetConfigApiTokenAuthEnabled(enabled bool) {
 	s.configApiTokenAuthEnabled = enabled
 }
 
+func (s *WebconfigServer) TokenApiTokenAuthEnabled() bool {
+	return s.tokenApiTokenAuthEnabled
+}
+
+func (s *WebconfigServer) SetTokenApiTokenAuthEnabled(enabled bool) {
+	s.tokenApiTokenAuthEnabled = enabled
+}
+
 func (s *WebconfigServer) DeviceApiTokenAuthEnabled() bool {
 	return s.deviceApiTokenAuthEnabled
 }
@@ -796,6 +821,17 @@ func (s *WebconfigServer) SetKafkaProducerTopic(x string) {
 	s.kafkaProducerTopic = x
 }
 
+func (s *WebconfigServer) KafkaLoggingMode() string {
+	return s.kafkaLoggingMode
+}
+
+func (s *WebconfigServer) SetKafkaLoggingMode(mode string) {
+	if mode != KafkaLoggingModeOneLine && mode != KafkaLoggingModeTwoLine {
+		return
+	}
+	s.kafkaLoggingMode = mode
+}
+
 func (s *WebconfigServer) UpstreamProfilesEnabled() bool {
 	return s.upstreamProfilesEnabled
 }
@@ -973,6 +1009,11 @@ func (s *WebconfigServer) logRequestStarts(w http.ResponseWriter, r *http.Reques
 		mac = strings.ToUpper(mac)
 		fields["cpe_mac"] = mac
 	}
+	if r.Method == "POST" {
+		if subdocID, ok := params["subdoc_id"]; ok {
+			fields["subdoc_id"] = subdocID
+		}
+	}
 
 	xwriter := NewXResponseWriter(w, time.Now(), token, fields)
 
@@ -1145,8 +1186,11 @@ func GetResponseLogObjs(rbytes []byte) (interface{}, string) {
 	return itf, ""
 }
 
-func (s *WebconfigServer) ForwardKafkaMessage(kbytes []byte, m *common.EventMessage, fields log.Fields) {
-	tfields := common.CopyCoreLogFields(fields)
+func (s *WebconfigServer) ForwardKafkaMessage(kbytes []byte, m *common.EventMessage, fields log.Fields, logMessage string) {
+	tfields := kafkaProducerLogFields(fields)
+	if _, ok := tfields["kafka_operation"]; !ok {
+		tfields["kafka_operation"] = "producer_send"
+	}
 
 	bbytes, err := json.Marshal(m)
 	if err != nil {
@@ -1168,7 +1212,7 @@ func (s *WebconfigServer) ForwardKafkaMessage(kbytes []byte, m *common.EventMess
 			if m := s.Metrics(); m != nil {
 				m.ObserveKafkaProducerErr(s.KafkaProducerTopic(), -1)
 			}
-			tfields["logger"] = "kafkaproducer"
+			tfields["logger"] = s.KafkaProducerLogger()
 			tfields["error"] = r
 			log.WithFields(tfields).Warn("dropped: producer closed during shutdown")
 		}
@@ -1176,16 +1220,32 @@ func (s *WebconfigServer) ForwardKafkaMessage(kbytes []byte, m *common.EventMess
 
 	s.Input() <- outMessage
 
-	tfields["logger"] = "kafkaproducer"
+	tfields["logger"] = s.KafkaProducerLogger()
 	tfields["output_topic"] = outMessage.Topic
 	tfields["output_key"] = string(kbytes)
-	tfields["output_body"] = m
-	log.WithFields(tfields).Info("send")
+	log.WithFields(tfields).Info(logMessage + "; send")
+}
+
+func kafkaProducerLogFields(fields log.Fields) log.Fields {
+	const producerFieldLimit = 15
+	allowedFields := [...]string{
+		"app_name", "audit_id", "body", "cpe_mac", "subdoc_id",
+		"kafka_key", "topic", "cluster_name", "kafka_partition", "kafka_offset",
+		"message_length", "duration", "event_name", "rpt", "kafka_operation",
+	}
+	producerFields := make(log.Fields, producerFieldLimit)
+	for _, field := range allowedFields {
+		if value, ok := fields[field]; ok {
+			producerFields[field] = common.FilterLogFields(log.Fields{field: value})[field]
+		}
+	}
+	return producerFields
 }
 
 func (s *WebconfigServer) ForwardSuccessKafkaMessages(messages []common.EventMessage, fields log.Fields) {
-	tfields := common.CopyCoreLogFields(fields)
-	tfields["logger"] = "kafkaproducer"
+	tfields := kafkaProducerLogFields(fields)
+	tfields["logger"] = s.KafkaProducerLogger()
+	tfields["kafka_operation"] = "state_correction_send"
 	tfields["output_topic"] = s.KafkaProducerTopic()
 
 	for _, m := range messages {
@@ -1228,7 +1288,6 @@ func (s *WebconfigServer) ForwardSuccessKafkaMessages(messages []common.EventMes
 		}
 
 		tfields["output_key"] = mac
-		tfields["output_body"] = m
 		log.WithFields(tfields).Info("send")
 	}
 }
@@ -1288,7 +1347,8 @@ func (s *WebconfigServer) HandleKafkaProducerResults(ctx context.Context) {
 				continue
 			}
 			fields := make(log.Fields)
-			fields["logger"] = "kafkaproducer"
+			fields["logger"] = s.KafkaProducerLogger()
+			fields["kafka_operation"] = "producer_result"
 			fields["output_topic"] = success.Topic
 			fields["output_partition"] = success.Partition
 			fields["output_offset"] = success.Offset
@@ -1304,7 +1364,8 @@ func (s *WebconfigServer) HandleKafkaProducerResults(ctx context.Context) {
 				m.ObserveKafkaProducerErr(pErr.Msg.Topic, pErr.Msg.Partition)
 			}
 			fields := make(log.Fields)
-			fields["logger"] = "kafkaproducer"
+			fields["logger"] = s.KafkaProducerLogger()
+			fields["kafka_operation"] = "producer_result"
 			fields["output_topic"] = pErr.Msg.Topic
 			fields["output_partition"] = pErr.Msg.Partition
 			kbytes, err := pErr.Msg.Key.Encode()
@@ -1318,14 +1379,7 @@ func (s *WebconfigServer) HandleKafkaProducerResults(ctx context.Context) {
 			if err != nil {
 				log.WithFields(fields).Error(common.NewError(err))
 			} else {
-				var itf interface{}
-				err1 := json.Unmarshal(vbytes, &itf)
-				if err1 != nil {
-					log.WithFields(fields).Error(common.NewError(err1))
-					fields["output_body_text"] = base64.StdEncoding.EncodeToString(vbytes)
-				} else {
-					fields["output_body"] = itf
-				}
+				common.UpdateLogFields(fields, producerErrorPayloadFields(vbytes))
 			}
 
 			log.WithFields(fields).Error(pErr.Err)
@@ -1333,6 +1387,40 @@ func (s *WebconfigServer) HandleKafkaProducerResults(ctx context.Context) {
 	}
 }
 
+func (s *WebconfigServer) KafkaProducerLogger() string {
+	if s.KafkaLoggingMode() == KafkaLoggingModeTwoLine {
+		return "kafkaproducer"
+	}
+	return "kafka"
+}
+
+func producerErrorPayloadFields(payload []byte) log.Fields {
+	fields := log.Fields{
+		"output_body_bytes":     len(payload),
+		"output_body_truncated": false,
+	}
+	maxRawBytes := (maxKafkaProducerLogPayloadBytes / 4) * 3
+	if len(payload) <= maxRawBytes {
+		var body interface{}
+		if err := json.Unmarshal(payload, &body); err == nil {
+			fields["output_body"] = body
+			return fields
+		}
+	}
+
+	fields["output_body_text"], fields["output_body_truncated"] = boundedKafkaPayload(payload)
+	return fields
+}
+
+func boundedKafkaPayload(payload []byte) (string, bool) {
+	maxRawBytes := (maxKafkaProducerLogPayloadBytes / 4) * 3
+	truncated := len(payload) > maxRawBytes
+	if truncated {
+		payload = payload[:maxRawBytes]
+	}
+	return base64.StdEncoding.EncodeToString(payload), truncated
+}
+
 func (s *WebconfigServer) StopXpcTracer() {
 	sdkTraceProvider, ok := s.XpcTracer.OtelTracerProvider().(*sdktrace.TracerProvider)
 	if ok && sdkTraceProvider != nil {
diff --git a/http/webconfig_server_test.go b/http/webconfig_server_test.go
index 6b611d4..a60a68f 100644
--- a/http/webconfig_server_test.go
+++ b/http/webconfig_server_test.go
@@ -44,9 +44,109 @@ func TestApiTokenAuthSecureDefaults(t *testing.T) {
 	// be guarded by ApiMiddleware out of the box. Regression guard for f003.
 	assert.Assert(t, serverApiTokenAuthEnabledDefault, "server API token auth must default to enabled")
 	assert.Assert(t, !configApiTokenAuthEnabledDefault, "config API token auth must default to disabled")
+	assert.Assert(t, !tokenApiTokenAuthEnabledDefault, "token API token auth must default to disabled")
 	assert.Assert(t, deviceApiTokenAuthEnabledDefault, "device API token auth must default to enabled")
 }
 
+func TestPostSubDocumentRequestIncludesSubdocID(t *testing.T) {
+	server := NewWebconfigServer(sc, true)
+
+	req, err := http.NewRequest("POST", "/api/v1/device/001122334455/document/lan", nil)
+	assert.NilError(t, err)
+	req = mux.SetURLVars(req, map[string]string{
+		"mac":       "001122334455",
+		"subdoc_id": "lan",
+	})
+	xw := server.logRequestStarts(httptest.NewRecorder(), req)
+	assert.Equal(t, xw.Audit()["subdoc_id"], "lan")
+
+	getReq, err := http.NewRequest("GET", "/api/v1/device/001122334455/document/lan", nil)
+	assert.NilError(t, err)
+	getReq = mux.SetURLVars(getReq, map[string]string{
+		"mac":       "001122334455",
+		"subdoc_id": "lan",
+	})
+	getXw := server.logRequestStarts(httptest.NewRecorder(), getReq)
+	_, ok := getXw.Audit()["subdoc_id"]
+	assert.Assert(t, !ok)
+}
+
+func TestKafkaLoggingModeDefaultsToOneLine(t *testing.T) {
+	server := NewWebconfigServer(sc, true)
+	assert.Equal(t, server.KafkaLoggingMode(), KafkaLoggingModeOneLine)
+
+	server.SetKafkaLoggingMode(KafkaLoggingModeTwoLine)
+	assert.Equal(t, server.KafkaLoggingMode(), KafkaLoggingModeTwoLine)
+	server.SetKafkaLoggingMode("invalid")
+	assert.Equal(t, server.KafkaLoggingMode(), KafkaLoggingModeTwoLine)
+}
+
+func TestKafkaProducerLoggerFollowsLoggingMode(t *testing.T) {
+	server := NewWebconfigServer(sc, true)
+	assert.Equal(t, server.KafkaProducerLogger(), "kafka")
+
+	server.SetKafkaLoggingMode(KafkaLoggingModeTwoLine)
+	assert.Equal(t, server.KafkaProducerLogger(), "kafkaproducer")
+}
+
+func TestKafkaProducerLogFieldsAllowlist(t *testing.T) {
+	fields := kafkaProducerLogFields(log.Fields{
+		"app_name":      "webconfig",
+		"audit_id":      "audit",
+		"cpe_mac":       "001122334455",
+		"subdoc_id":     "lan",
+		"event_name":    "mqtt-get",
+		"header":        map[string]string{"Authorization": "secret", "X-Test": "value"},
+		"authorization": "secret",
+	})
+
+	_, hasHeader := fields["header"]
+	_, hasAuthorization := fields["authorization"]
+	assert.Assert(t, !hasHeader)
+	assert.Assert(t, !hasAuthorization)
+	assert.Equal(t, fields["subdoc_id"], "lan")
+	assert.Equal(t, fields["event_name"], "mqtt-get")
+	assert.Equal(t, fields["kafka_operation"], nil)
+}
+
+func TestBoundedKafkaPayload(t *testing.T) {
+	payload := make([]byte, maxKafkaProducerLogPayloadBytes)
+	bounded, truncated := boundedKafkaPayload(payload)
+	assert.Equal(t, len(bounded), maxKafkaProducerLogPayloadBytes)
+	assert.Assert(t, truncated)
+	_, err := base64.StdEncoding.DecodeString(bounded)
+	assert.NilError(t, err)
+
+	payload = append(payload, 'x')
+	bounded, truncated = boundedKafkaPayload(payload)
+	assert.Equal(t, len(bounded), maxKafkaProducerLogPayloadBytes)
+	assert.Assert(t, truncated)
+	_, err = base64.StdEncoding.DecodeString(bounded)
+	assert.NilError(t, err)
+
+	payload = payload[:(maxKafkaProducerLogPayloadBytes/4)*3]
+	bounded, truncated = boundedKafkaPayload(payload)
+	assert.Assert(t, !truncated)
+	assert.Equal(t, len(bounded), maxKafkaProducerLogPayloadBytes)
+}
+
+func TestProducerErrorPayloadFields(t *testing.T) {
+	fields := producerErrorPayloadFields([]byte(`{"status":"failed"}`))
+	assert.Equal(t, fields["output_body"].(map[string]interface{})["status"], "failed")
+	_, hasText := fields["output_body_text"]
+	assert.Assert(t, !hasText)
+	assert.Assert(t, !fields["output_body_truncated"].(bool))
+
+	fields = producerErrorPayloadFields([]byte("not-json"))
+	_, hasBody := fields["output_body"]
+	assert.Assert(t, !hasBody)
+	_, hasText = fields["output_body_text"]
+	assert.Assert(t, hasText)
+
+	fields = producerErrorPayloadFields(make([]byte, maxKafkaProducerLogPayloadBytes))
+	assert.Assert(t, fields["output_body_truncated"].(bool))
+}
+
 func TestConfigEndpointRemainsUnauthenticatedByDefault(t *testing.T) {
 	server := NewWebconfigServer(sc, true)
 	assert.Assert(t, !server.ConfigApiTokenAuthEnabled())
@@ -70,6 +170,54 @@ func TestConfigEndpointRequiresApiTokenWhenEnabled(t *testing.T) {
 	assert.Equal(t, res.StatusCode, http.StatusUnauthorized)
 }
 
+func TestTokenEndpointRemainsUnauthenticatedWhenDisabled(t *testing.T) {
+	server := NewWebconfigServer(sc, true)
+	server.SetTokenApiEnabled(true)
+	assert.Assert(t, !server.TokenApiTokenAuthEnabled())
+	router := server.GetRouter(false)
+
+	req, err := http.NewRequest("POST", "/api/v1/token", strings.NewReader("not-json"))
+	assert.NilError(t, err)
+	res := ExecuteRequest(req, router).Result()
+	assert.Equal(t, res.StatusCode, http.StatusInternalServerError)
+}
+
+func TestTokenEndpointRequiresApiTokenWhenEnabled(t *testing.T) {
+	server := NewWebconfigServer(sc, true)
+	server.SetTokenApiEnabled(true)
+	server.SetTokenApiTokenAuthEnabled(true)
+	router := server.GetRouter(false)
+
+	req, err := http.NewRequest("POST", "/api/v1/token", strings.NewReader("not-json"))
+	assert.NilError(t, err)
+	res := ExecuteRequest(req, router).Result()
+	assert.Equal(t, res.StatusCode, http.StatusUnauthorized)
+}
+
+func TestTokenEndpointUsesTestingMiddlewareInTestRouter(t *testing.T) {
+	server := NewWebconfigServer(sc, true)
+	server.SetTokenApiEnabled(true)
+	server.SetTokenApiTokenAuthEnabled(true)
+	router := server.GetRouter(true)
+
+	req, err := http.NewRequest("POST", "/api/v1/token", strings.NewReader("not-json"))
+	assert.NilError(t, err)
+	res := ExecuteRequest(req, router).Result()
+	assert.Equal(t, res.StatusCode, http.StatusInternalServerError)
+}
+
+func TestConfigEndpointUsesTestingMiddlewareInTestRouter(t *testing.T) {
+	server := NewWebconfigServer(sc, true)
+	server.SetConfigApiTokenAuthEnabled(true)
+	router := server.GetRouter(true)
+
+	req, err := http.NewRequest("GET", "/config", nil)
+	assert.NilError(t, err)
+	req.Header.Set("Authorization", "Bearer test-token")
+	res := ExecuteRequest(req, router).Result()
+	assert.Equal(t, res.StatusCode, http.StatusOK)
+}
+
 func TestApiMiddlewareSuppressesConfigRequestLogs(t *testing.T) {
 	server := NewWebconfigServer(sc, true)
 	server.SetConfigApiTokenAuthEnabled(true)
@@ -605,6 +753,14 @@ func TestWebconfigServerSetterGetter(t *testing.T) {
 	server.SetConfigApiTokenAuthEnabled(enabled)
 	assert.Equal(t, server.ConfigApiTokenAuthEnabled(), enabled)
 
+	// token api token auth
+	enabled = true
+	server.SetTokenApiTokenAuthEnabled(enabled)
+	assert.Equal(t, server.TokenApiTokenAuthEnabled(), enabled)
+	enabled = false
+	server.SetTokenApiTokenAuthEnabled(enabled)
+	assert.Equal(t, server.TokenApiTokenAuthEnabled(), enabled)
+
 	// device api token auth
 	enabled = true
 	server.SetDeviceApiTokenAuthEnabled(enabled)
diff --git a/kafka/consumer.go b/kafka/consumer.go
index adb95e6..90eea28 100644
--- a/kafka/consumer.go
+++ b/kafka/consumer.go
@@ -46,6 +46,30 @@ type Consumer struct {
 	topicPartitionsMap         map[string][]int32
 }
 
+func addKafkaEventLogFields(fields log.Fields, eventName, rptHeaderValue string) {
+	if eventName != "webpa-state" {
+		fields["event_name"] = eventName
+	}
+	if rptHeaderValue != "" {
+		fields["rpt"] = rptHeaderValue
+	}
+}
+
+const maxKafkaMessageLogPayloadBytes = 4096
+
+func boundedKafkaMessage(payload []byte) (string, bool) {
+	maxRawBytes := (maxKafkaMessageLogPayloadBytes / 4) * 3
+	truncated := len(payload) > maxRawBytes
+	if truncated {
+		payload = payload[:maxRawBytes]
+	}
+	return base64.StdEncoding.EncodeToString(payload), truncated
+}
+
+func shouldLogConsumerSuccess(producerEnabled bool, message *common.EventMessage, loggingMode string) bool {
+	return !producerEnabled || message == nil || loggingMode == wchttp.KafkaLoggingModeTwoLine
+}
+
 func NewConsumer(s *wchttp.WebconfigServer, ratelimitMessagesPerSecond int, m *common.AppMetrics, clusterName string, offsetEnum int64, topicPartitionsMap map[string][]int32) *Consumer {
 	return &Consumer{
 		WebconfigServer:            s,
@@ -210,6 +234,7 @@ func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim saram
 			messageLength := len(message.Value)
 			fields := log.Fields{
 				"logger":          "kafka",
+				"kafka_operation": "consumer_process",
 				"app_name":        c.AppName(),
 				"kafka_lag":       lag,
 				"kafka_key":       kafkaKey,
@@ -244,8 +269,7 @@ func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim saram
 			session.MarkMessage(message, "")
 			duration := int(time.Since(start).Nanoseconds() / 1000000)
 			fields["duration"] = duration
-			fields["event_name"] = eventName
-			fields["rpt"] = rptHeaderValue
+			addKafkaEventLogFields(fields, eventName, rptHeaderValue)
 
 			forwardMessage := false
 			if err != nil {
@@ -255,12 +279,15 @@ func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim saram
 					log.WithFields(fields).Trace("pending")
 				} else {
 					fields["error"] = err.Error()
-					fields["kafka_message"] = base64.StdEncoding.EncodeToString(message.Value)
+					fields["kafka_message"], fields["kafka_message_truncated"] = boundedKafkaMessage(message.Value)
+					fields["kafka_message_bytes"] = len(message.Value)
 					log.WithFields(fields).Error("errors")
 				}
 			} else {
 				forwardMessage = true
-				log.WithFields(fields).Info(logMessage)
+				if shouldLogConsumerSuccess(c.KafkaProducerEnabled(), m, c.KafkaLoggingMode()) {
+					log.WithFields(fields).Info(logMessage)
+				}
 			}
 
 			// build metrics dimensions and update metrics
@@ -281,7 +308,8 @@ func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim saram
 			}
 
 			if c.KafkaProducerEnabled() && m != nil && forwardMessage {
-				c.ForwardKafkaMessage(message.Key, m, fields)
+				fields["kafka_operation"] = "producer_send"
+				c.ForwardKafkaMessage(message.Key, m, fields, logMessage)
 				if len(m.Reports) == 0 {
 					if m.HttpStatusCode != nil && *m.HttpStatusCode == http.StatusNotModified && len(updatedSubdocIds) > 0 {
 						// build a root/success message
@@ -294,7 +322,8 @@ func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim saram
 								TransactionUuid:   m.TransactionUuid,
 								Version:           m.Version,
 							}
-							c.ForwardKafkaMessage(message.Key, em, fields)
+							fields["kafka_operation"] = "state_correction_send"
+							c.ForwardKafkaMessage(message.Key, em, fields, logMessage)
 						}
 					}
 				}
@@ -303,7 +332,6 @@ func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim saram
 			return nil
 		}
 	}
-	return nil
 }
 
 func (c *Consumer) AppName() string {
diff --git a/kafka/consumer_group_test.go b/kafka/consumer_group_test.go
index d720eb2..c1fcd78 100644
--- a/kafka/consumer_group_test.go
+++ b/kafka/consumer_group_test.go
@@ -14,17 +14,90 @@
 * limitations under the License.
 *
 * SPDX-License-Identifier: Apache-2.0
-*/
+ */
 package kafka
 
 import (
+	"encoding/base64"
 	"testing"
 	"time"
 
 	"github.com/IBM/sarama"
+	"github.com/rdkcentral/webconfig/common"
+	wchttp "github.com/rdkcentral/webconfig/http"
+	log "github.com/sirupsen/logrus"
 	"gotest.tools/assert"
 )
 
+func TestKafkaSuccessLoggingMode(t *testing.T) {
+	message := &common.EventMessage{}
+	tests := []struct {
+		name            string
+		producerEnabled bool
+		message         *common.EventMessage
+		loggingMode     string
+		want            bool
+	}{
+		{name: "producer disabled", want: true},
+		{name: "one line", producerEnabled: true, message: message, loggingMode: wchttp.KafkaLoggingModeOneLine, want: false},
+		{name: "two line", producerEnabled: true, message: message, loggingMode: wchttp.KafkaLoggingModeTwoLine, want: true},
+		{name: "no message", producerEnabled: true, loggingMode: wchttp.KafkaLoggingModeOneLine, want: true},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			assert.Equal(t, shouldLogConsumerSuccess(test.producerEnabled, test.message, test.loggingMode), test.want)
+		})
+	}
+}
+
+func TestBoundedKafkaMessage(t *testing.T) {
+	payload := make([]byte, maxKafkaMessageLogPayloadBytes+1)
+	encoded, truncated := boundedKafkaMessage(payload)
+	assert.Assert(t, truncated)
+	assert.Equal(t, len(encoded), maxKafkaMessageLogPayloadBytes)
+	_, err := base64.StdEncoding.DecodeString(encoded)
+	assert.NilError(t, err)
+
+	payload = payload[:(maxKafkaMessageLogPayloadBytes/4)*3]
+	encoded, truncated = boundedKafkaMessage(payload)
+	assert.Assert(t, !truncated)
+	assert.Equal(t, len(encoded), maxKafkaMessageLogPayloadBytes)
+}
+
+func TestKafkaEventLogFields(t *testing.T) {
+	tests := []struct {
+		name      string
+		eventName string
+		rpt       string
+		wantEvent string
+		wantRPT   string
+		hasEvent  bool
+		hasRPT    bool
+	}{
+		{name: "webpa state default", eventName: "webpa-state"},
+		{name: "mqtt get", eventName: "mqtt-get", rpt: "x/fr/webconfig/get", wantEvent: "mqtt-get", wantRPT: "x/fr/webconfig/get", hasEvent: true, hasRPT: true},
+		{name: "mqtt state", eventName: "mqtt-state", rpt: "x/fr/webconfig/poke", wantEvent: "mqtt-state", wantRPT: "x/fr/webconfig/poke", hasEvent: true, hasRPT: true},
+		{name: "unknown event without rpt", eventName: "unknown-no-rpt", wantEvent: "unknown-no-rpt", hasEvent: true},
+		{name: "unknown event with rpt", eventName: "unknown-rpt", rpt: "indigo", wantEvent: "unknown-rpt", wantRPT: "indigo", hasEvent: true, hasRPT: true},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			fields := log.Fields{}
+			addKafkaEventLogFields(fields, test.eventName, test.rpt)
+			value, hasEvent := fields["event_name"]
+			assert.Equal(t, hasEvent, test.hasEvent)
+			if test.hasEvent {
+				assert.Equal(t, value, test.wantEvent)
+			}
+			value, hasRPT := fields["rpt"]
+			assert.Equal(t, hasRPT, test.hasRPT)
+			if test.hasRPT {
+				assert.Equal(t, value, test.wantRPT)
+			}
+		})
+	}
+}
+
 func TestGetEventName(t *testing.T) {
 	// ==== mqtt-get ====
 	rheader := &sarama.RecordHeader{
diff --git a/security/token.go b/security/token.go
index 7420a63..42b160c 100644
--- a/security/token.go
+++ b/security/token.go
@@ -342,16 +342,37 @@ func VerifyToken(decodeKeys map[string]*rsa.PublicKey, validKids []string, requi
 		return false, "", trust, common.NewError(err)
 	}
 
-	// check capabilities, if requiredCapabilities is nonempty
-	// this must run only after the token has been cryptographically verified
+	if len(vargs) > 1 {
+		mac := vargs[1]
+		// mac must match
+		isMatched := false
+		if macitf, ok := claims["mac"]; ok {
+			if macstr, ok := macitf.(string); ok {
+				if strings.ToLower(mac) == strings.ToLower(macstr) {
+					isMatched = true
+				} else {
+					return false, "", trust, common.NewError(fmt.Errorf("mac in token(%v) does not match mac in claims(%v)", mac, macstr))
+				}
+			}
+		}
+		if !isMatched {
+			return false, "", trust, common.NewError(fmt.Errorf("mac in token(%v) does not match claims=%v", mac, claims))
+		}
+	}
+
+	// Check capabilities after signature and route binding validation. Entries
+	// are untrusted claim values and must be strings before comparison.
 	if len(requiredCapabilities) > 0 {
 		isCapable := false
 		if capitfs, ok := claims["capabilities"]; ok {
-			capvalues, ok1 := capitfs.([]interface{})
-			if ok1 {
+			if capvalues, ok := capitfs.([]interface{}); ok {
 				for _, capvalue := range capvalues {
-					for _, rc := range requiredCapabilities {
-						if rc == capvalue {
+					capability, ok := capvalue.(string)
+					if !ok {
+						continue
+					}
+					for _, requiredCapability := range requiredCapabilities {
+						if requiredCapability == capability {
 							isCapable = true
 							break
 						}
@@ -367,24 +388,6 @@ func VerifyToken(decodeKeys map[string]*rsa.PublicKey, validKids []string, requi
 		}
 	}
 
-	if len(vargs) > 1 {
-		mac := vargs[1]
-		// mac must match
-		isMatched := false
-		if macitf, ok := claims["mac"]; ok {
-			if macstr, ok := macitf.(string); ok {
-				if strings.ToLower(mac) == strings.ToLower(macstr) {
-					isMatched = true
-				} else {
-					return false, "", trust, common.NewError(fmt.Errorf("mac in token(%v) does not match mac in claims(%v)", mac, macstr))
-				}
-			}
-		}
-		if !isMatched {
-			return false, "", trust, common.NewError(fmt.Errorf("mac in token(%v) does not match claims=%v", mac, claims))
-		}
-	}
-
 	// parse partner
 	partner := "comcast"
 	if itf, ok := claims["partner-id"]; ok {
diff --git a/security/token_test.go b/security/token_test.go
index ae7506b..728c8c6 100644
--- a/security/token_test.go
+++ b/security/token_test.go
@@ -184,6 +184,50 @@ func TestVerifyTokenEnforcesCpeCapabilitiesWhenConfigured(t *testing.T) {
 	assert.Assert(t, errors.Is(err, common.ErrNoCapabilities))
 }
 
+func TestVerifyTokenRejectsNonStringCapabilityWithoutPanic(t *testing.T) {
+	if tokenManager == nil {
+		t.Skip("webconfig.jwt.enabled = false")
+	}
+
+	cpeMac := util.GenerateRandomCpeMac()
+	claims := jwt.MapClaims{
+		"mac":          strings.ToLower(cpeMac),
+		"capabilities": []interface{}{map[string]interface{}{"nested": true}},
+		"exp":          time.Now().Add(time.Hour).Unix(),
+	}
+	token := jwt.NewWithClaims(jwt.GetSigningMethod("RS256"), claims)
+	token.Header["kid"] = EncodingKeyId
+	tokenString, err := token.SignedString(tokenManager.encodeKey)
+	assert.NilError(t, err)
+
+	ok, _, _, verifyErr := VerifyToken(tokenManager.decodeKeys, tokenManager.cpeKids, []string{"cpe:config:read"}, tokenString, cpeMac)
+	assert.Assert(t, !ok)
+	assert.Assert(t, errors.Is(verifyErr, common.ErrNoCapabilities))
+}
+
+func TestVerifyTokenChecksMacBeforeCapabilities(t *testing.T) {
+	if tokenManager == nil {
+		t.Skip("webconfig.jwt.enabled = false")
+	}
+
+	tokenMac := util.GenerateRandomCpeMac()
+	requestMac := util.GenerateRandomCpeMac()
+	claims := jwt.MapClaims{
+		"mac":          strings.ToLower(tokenMac),
+		"capabilities": []string{"cpe:other"},
+		"exp":          time.Now().Add(time.Hour).Unix(),
+	}
+	token := jwt.NewWithClaims(jwt.GetSigningMethod("RS256"), claims)
+	token.Header["kid"] = EncodingKeyId
+	tokenString, err := token.SignedString(tokenManager.encodeKey)
+	assert.NilError(t, err)
+
+	ok, _, _, verifyErr := VerifyToken(tokenManager.decodeKeys, tokenManager.cpeKids, []string{"cpe:config:read"}, tokenString, requestMac)
+	assert.Assert(t, !ok)
+	assert.Assert(t, !errors.Is(verifyErr, common.ErrNoCapabilities))
+	assert.ErrorContains(t, verifyErr, "does not match")
+}
+
 // XPC-42727 an expired token must fail as an authentication error, never
 // misclassified as ErrNoCapabilities, even when its (untrusted) claims lack
 // the required capability