Description

I'm attempting to use the latest features in Mirror Maker 2 (2.7.1) to replicate data to another Event Hubs and am getting an error when running mirror maker.

How to reproduce

Create a mirror maker configuration file using 2 Event Hubs.

clusters = A, B

A.bootstrap.servers=<source event hubs>.servicebus.windows.net:9093
A.security.protocol=SASL_SSL
A.sasl.mechanism=PLAIN
A.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="Endpoint=sb://<source event hubs>.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=ZZZZ";
B.bootstrap.servers=<destination event hubs>.servicebus.windows.net:9093
B.security.protocol=SASL_SSL
B.sasl.mechanism=PLAIN
B.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="Endpoint=sb://<destination event hubs>.servicebus.windows.net/;SharedAccessKeyName=ehrepl;SharedAccessKey=ZZZZ";

A->B.enabled = true

A->B.topics = .*

replication.policy.separator =
source.cluster.alias = 
target.cluster.alias =

############################# Internal Topic Settings  #############################
# The replication factor for mm2 internal topics "heartbeats", "B.checkpoints.internal" and
# "mm2-offset-syncs.B.internal"
# For anything other than development testing, a value greater than 1 is recommended to ensure availability such as 3.
checkpoints.topic.replication.factor=1
heartbeats.topic.replication.factor=1
offset-syncs.topic.replication.factor=1

# The replication factor for connect internal topics "mm2-configs.B.internal", "mm2-offsets.B.internal" and
# "mm2-status.B.internal"
# For anything other than development testing, a value greater than 1 is recommended to ensure availability such as 3.
offset.storage.replication.factor=1
status.storage.replication.factor=1
config.storage.replication.factor=1

# customize as needed
# replication.policy.separator = _
# sync.topic.acls.enabled = false
# emit.heartbeats.interval.seconds = 5

Run connect-mirror-maker.sh <config file>

The Kafka Admin Client fails to be able to fully authenticate with Event Hubs with the following error

[2021-10-27 11:57:20,281] INFO Creating Kafka admin client (org.apache.kafka.connect.util.ConnectUtils:49)
[2021-10-27 11:57:20,285] INFO AdminClientConfig values:
    bootstrap.servers = XXXX.servicebus.windows.net:9093]
    client.dns.lookup = use_all_dns_ips
    client.id =
    connections.max.idle.ms = 300000
    default.api.timeout.ms = 60000
    metadata.max.age.ms = 300000
    metric.reporters = []
    metrics.num.samples = 2
    metrics.recording.level = INFO
    metrics.sample.window.ms = 30000
    receive.buffer.bytes = 65536
    reconnect.backoff.max.ms = 1000
    reconnect.backoff.ms = 50
    request.timeout.ms = 30000
    retries = 2147483647
    retry.backoff.ms = 100
    sasl.client.callback.handler.class = null
    sasl.jaas.config = [hidden]
    sasl.kerberos.kinit.cmd = /usr/bin/kinit
    sasl.kerberos.min.time.before.relogin = 60000
    sasl.kerberos.service.name = null
    sasl.kerberos.ticket.renew.jitter = 0.05
    sasl.kerberos.ticket.renew.window.factor = 0.8
    sasl.login.callback.handler.class = null
    sasl.login.class = null
    sasl.login.refresh.buffer.seconds = 300
    sasl.login.refresh.min.period.seconds = 60
    sasl.login.refresh.window.factor = 0.8
    sasl.login.refresh.window.jitter = 0.05
    sasl.mechanism = PLAIN
    security.protocol = SASL_SSL
    security.providers = null
    send.buffer.bytes = 131072
    socket.connection.setup.timeout.max.ms = 127000
    socket.connection.setup.timeout.ms = 10000
    ssl.cipher.suites = null
    ssl.enabled.protocols = [TLSv1.2]
    ssl.endpoint.identification.algorithm = https
    ssl.engine.factory.class = null
    ssl.key.password = null
    ssl.keymanager.algorithm = SunX509
    ssl.keystore.certificate.chain = null
    ssl.keystore.key = null
    ssl.keystore.location = null
    ssl.keystore.password = null
    ssl.keystore.type = JKS
    ssl.protocol = TLSv1.2
    ssl.provider = null
    ssl.secure.random.implementation = null
    ssl.trustmanager.algorithm = PKIX
    ssl.truststore.certificates = null
    ssl.truststore.location = null
    ssl.truststore.password = null
    ssl.truststore.type = JKS
 (org.apache.kafka.clients.admin.AdminClientConfig:361)
[2021-10-27 11:57:20,488] INFO Successfully logged in. (org.apache.kafka.common.security.authenticator.AbstractLogin:61)
[2021-10-27 11:57:20,753] WARN The configuration 'producer.sasl.jaas.config' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig:369)
[2021-10-27 11:57:20,753] WARN The configuration 'producer.bootstrap.servers' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig:369)
[2021-10-27 11:57:20,753] WARN The configuration 'group.id' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig:369)
[2021-10-27 11:57:20,753] WARN The configuration 'consumer.sasl.mechanism' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig:369)
[2021-10-27 11:57:20,754] WARN The configuration 'admin.security.protocol' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig:369)
[2021-10-27 11:57:20,754] WARN The configuration 'sasl.jaas.config' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig:369)
[2021-10-27 11:57:20,754] WARN The configuration 'status.storage.replication.factor' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig:369)
[2021-10-27 11:57:20,754] WARN The configuration 'consumer.security.protocol' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig:369)
[2021-10-27 11:57:20,754] WARN The configuration 'admin.sasl.mechanism' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig:369)
[2021-10-27 11:57:20,755] WARN The configuration 'offset.storage.topic' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig:369)
[2021-10-27 11:57:20,755] WARN The configuration 'value.converter' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig:369)
[2021-10-27 11:57:20,755] WARN The configuration 'key.converter' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig:369)
[2021-10-27 11:57:20,755] WARN The configuration 'consumer.sasl.jaas.config' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig:369)
[2021-10-27 11:57:20,755] WARN The configuration 'producer.security.protocol' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig:369)
[2021-10-27 11:57:20,755] WARN The configuration 'config.storage.topic' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig:369)
[2021-10-27 11:57:20,755] WARN The configuration 'status.storage.topic' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig:369)
[2021-10-27 11:57:20,755] WARN The configuration 'header.converter' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig:369)
[2021-10-27 11:57:20,756] WARN The configuration 'consumer.bootstrap.servers' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig:369)
[2021-10-27 11:57:20,756] WARN The configuration 'producer.sasl.mechanism' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig:369)
[2021-10-27 11:57:20,756] WARN The configuration 'config.storage.replication.factor' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig:369)
[2021-10-27 11:57:20,756] WARN The configuration 'offset.storage.replication.factor' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig:369)
[2021-10-27 11:57:20,757] WARN The configuration 'admin.sasl.jaas.config' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig:369)
[2021-10-27 11:57:20,757] WARN The configuration 'admin.bootstrap.servers' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig:369)
[2021-10-27 11:57:20,758] INFO Kafka version: 2.7.1 (org.apache.kafka.common.utils.AppInfoParser:119)
[2021-10-27 11:57:20,758] INFO Kafka commitId: 61dbce85d0d41457 (org.apache.kafka.common.utils.AppInfoParser:120)
[2021-10-27 11:57:20,758] INFO Kafka startTimeMs: 1635335840757 (org.apache.kafka.common.utils.AppInfoParser:121)
[2021-10-27 11:57:22,300] INFO [AdminClient clientId=adminclient-1] Failed authentication with XXXX.servicebus.windows.net/52.167.109.X (Invalid SASL mechanism response, server may be expecting a different protocol) (org.apache.kafka.common.network.Selector:616)
[2021-10-27 11:57:22,307] ERROR [AdminClient clientId=adminclient-1] Connection to node -1 (XXXX.servicebus.windows.net/52.167.109.X:9093) failed authentication due to: Invalid SASL mechanism response, server may be expecting a different protocol (org.apache.kafka.clients.NetworkClient:771)
[2021-10-27 11:57:22,309] WARN [AdminClient clientId=adminclient-1] Metadata update failed due to authentication error (org.apache.kafka.clients.admin.internals.AdminMetadataManager:232)
org.apache.kafka.common.errors.IllegalSaslStateException: Invalid SASL mechanism response, server may be expecting a different protocol
Caused by: org.apache.kafka.common.protocol.types.SchemaException: Error reading field 'auth_bytes': Bytes size -1 cannot be negative
    at org.apache.kafka.common.protocol.types.Schema.read(Schema.java:118)
    at org.apache.kafka.common.protocol.ApiKeys.parseResponse(ApiKeys.java:379)
    at org.apache.kafka.clients.NetworkClient.parseStructMaybeUpdateThrottleTimeMetrics(NetworkClient.java:745)
    at org.apache.kafka.clients.NetworkClient.parseResponse(NetworkClient.java:732)
    at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.receiveKafkaResponse(SaslClientAuthenticator.java:564)
    at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.receiveToken(SaslClientAuthenticator.java:498)
    at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.authenticate(SaslClientAuthenticator.java:300)
    at org.apache.kafka.common.network.KafkaChannel.prepare(KafkaChannel.java:176)
    at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:543)
    at org.apache.kafka.common.network.Selector.poll(Selector.java:481)
    at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:563)
    at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.processRequests(KafkaAdminClient.java:1329)
    at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run(KafkaAdminClient.java:1260)
    at java.lang.Thread.run(Thread.java:748)

Has it worked previously?

Checklist

IMPORTANT: We will close issues where the checklist has not been completed or where adequate information has not been provided.

Please provide the relevant information for the following items:

[ ] SDK (include version info): <REPLACE with e.g., kafka-python/Java SDK/confluent-kafka-dotnet with version info>
[ ] Sample you're having trouble with: <REPLACE with e.g., Java quickstart>
[x] If using Apache Kafka Java clients or a framework that uses Apache Kafka Java clients, version: <REPLACE with e.g., 1.1.0>
[x] Kafka client configuration: <REPLACE with e.g., auto.reset.offset=earliest, ..> (do not include your connection string or SAS Key)
[ ] Namespace and EventHub/topic name
[ ] Consumer or producer failure <REPLACE with e.g., Consumer failure>
[ ] Timestamps in UTC <REPLACE with e.g., Nov 7 2018 - 17:15:01 UTC>
[ ] group.id or client.id <REPLACE with e.g., group.id=cg-name>
[ ] Logs provided (with debug-level logging enabled if possible, e.g. log4j.rootLogger=DEBUG) or exception call stack
[ ] Standalone repro <REPLACE with e.g., Willing/able to send scenario to repro issue>
[ ] Operating system: <REPLACE with e.g., Ubuntu 16.04.5 (x64) LTS>
[ ] Critical issue

If this is a question on basic functionality, please verify the following:

x ] Port 9093 should not be blocked by firewall ("broker cannot be found" errors)
[x] Pinging FQDN should return cluster DNS resolution (e.g. $ ping namespace.servicebus.windows.net returns ~ ns-eh2-prod-am3-516.cloudapp.net [13.69.64.0])
[x] Namespace should be either Standard or Dedicated tier, not Basic (TopicAuthorization errors)

Azure / azure-event-hubs-for-kafka

Caused by: org.apache.kafka.common.protocol.types.SchemaException: Error reading field 'auth_bytes': Bytes size -1 cannot be negative #169

Description

How to reproduce

Has it worked previously?

Checklist