[BUG] - Kafka authentication "check_hostname" false is not beign applied.

[messerzen]

Mage version

0.9.70

Describe the bug

I'm trying to authenticate Kafka within a streaming pipeline like this:

connector_type: kafka
# bootstrap_server: "{{ env_var('DATAPLATFORM01_KAFKA_SERVERS') }}" 
topic: visibilidade-direto.transient-ingestion
consumer_group: mage-to-hwc-kafka-listener
include_metadata: false
api_version: 0.10.2

# Uncomment the config below to use SSL config
# security_protocol: "SSL"
# ssl_config:
#   cafile: "/etc/kafka/ssl/ca.pem"
#   # certfile: "/etc/kafka/ssl/cert.pem"
#   # keyfile: "/etc/kafka/ssl/key.pem"
#   # password: "{{ env_var('KAFKA_PASSWORD') }}"
#   check_hostname: false

# Uncomment the config below to use SASL_SSL config
security_protocol: "SASL_SSL"
sasl_config:
  mechanism: "PLAIN"
  username: "{{ env_var('KAFKA_USERNAME') }}"
  password: "{{ env_var('KAFKA_PASSWORD') }}"
ssl_config:
  cafile: "/etc/kafka/ssl/ca.pem"
  certfile: "/etc/kafka/ssl/cert.pem"
  keyfile: "/etc/kafka/ssl/key.pem"
  password: "{{ env_var('KAFKA_PASSWORD') }}"
  check_hostname: false

But when running the pipeline it's not matching the ssl certificate.

When trying to verify the handshake in the pod it works.
openssl s_client -connect <broker_ip> -CAfile /etc/kafka/ssl/ca.pem -cert /etc/kafka/ssl/cert.pem -key /etc/kafka/ssl/key.pem -showcerts

To reproduce

No response

Expected behavior

No response

Screenshots

No response

Operating system

No response

Additional context

No response

[messerzen]

I've made a test with data loader in mage, I it worked

    from mage_ai.data_preparation.decorators import data_loader
if 'test' not in globals():
    from mage_ai.data_preparation.decorators import test

from confluent_kafka import Consumer, KafkaError
import os

def consume():
    
    conf = {
    'bootstrap.servers': os.getenv('DATAPLATFORM01_KAFKA_SERVERS'),
    'group.id': 'mage-to-hwc-kafka-listener',
    'auto.offset.reset': 'earliest',
    'debug': 'conf'
    }   

    #if os.getenv('KAFKA_USERNAME') and os.getenv('KAFKA_PASSWORD'):
    conf['security.protocol'] = 'SASL_SSL'
    conf['sasl.mechanisms'] = 'PLAIN'
    conf['sasl.username'] = os.getenv('KAFKA_USERNAME')
    conf['sasl.password'] = os.getenv('KAFKA_PASSWORD')
    conf['ssl.ca.location'] = '/etc/kafka/ssl/ca.pem'
    conf['ssl.certificate.location'] = '/etc/kafka/ssl/cert.pem'
    conf['ssl.key.location'] = '/etc/kafka/ssl/key.pem'
    conf['ssl.key.password'] = "password"
    conf['ssl.endpoint.identification.algorithm']= 'none'

    consumer = Consumer(conf)
    consumer.subscribe(['visibilidade-direto.transient-ingestion'])

    try:
        print("Starting consuming")
        while True:
            msg = consumer.poll(timeout=10.0)
            if msg is None:
                continue
            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    continue
                else:
                    print(msg.error())
                    break
            print('Received message: {}'.format(msg.value().decode('utf-8')))
    finally:
        print("Consuming finishing)")
        consumer.close()

@data_loader
def load_data(*args, **kwargs):
    """
    Template code for loading data from any source.

    Returns:
        Anything (e.g. data frame, dictionary, array, int, str, etc.)
    """
    # Specify your data loading logic here


    
    consume()

So I think the problem is check_hostname: false not beign applied.