search

strimzi / strimzi-kafka-operator

Apache Kafka® running on Kubernetes
https://strimzi.io/
Apache License 2.0
4.78k stars 1.28k forks source link

[Question] Single MirrorMaker2 resource accessing to every $ConnectionString defined as secret. #3714

Closed yuwtennis closed 3 years ago

yuwtennis commented 3 years ago

Hello.

I am facing a problem with Mirror Maker 2 . Appreciate if I could get some help to work it out.

Environment

Below are the software versions I use for my architecture.

Software Version
Strimzi 0.17.0
Kafka 2.4.0

Architecture

What I am trying to do is mirror date to other AKS cluster using Azure eventhub. Fow is below. I am using three MirrorMaker2s due to eventhub quota .

| AKS1------------------------- | Azure EventHub----- | AKS2 --------------------------- | Kafka cluster A -> MM2 1 -------> eventhub1 ------> MM2 1 -> Kafka cluster B Kafka cluster A -> MM2 2 -------> eventhub2 ------> MM2 2 -> Kafka cluster B Kafka cluster A -> MM2 3 -------> eventhub3 ------> MM2 3 -> Kafka cluster B

Problem

I am facing below error in MIrror Maker 2 on AKS2 .
This error seems to be blocking mirror maker 2 to consume logs from eventhub.

2020-09-27 02:08:16,019 INFO [AdminClient clientId=adminclient-21533] Failed authentication with migration-kafka-topics-connect.servicebus.windows.net/40.79.186.34 (Invalid SASL mechanism response, server may be expecting a different protocol) (org.apache.kafka.common.network.Selector) [kafka-admin-client-thread | adminclient-21533]
2020-09-27 02:08:16,019 WARN [AdminClient clientId=adminclient-21533] Metadata update failed due to authentication error (org.apache.kafka.clients.admin.internals.AdminMetadataManager) [kafka-admin-client-thread | adminclient-21533]
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:110)
        at org.apache.kafka.common.protocol.ApiKeys.parseResponse(ApiKeys.java:314)
        at org.apache.kafka.clients.NetworkClient.parseStructMaybeUpdateThrottleTimeMetrics(NetworkClient.java:712)
        at org.apache.kafka.clients.NetworkClient.parseResponse(NetworkClient.java:702)
        at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.receiveKafkaResponse(SaslClientAuthenticator.java:514)
        at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.receiveToken(SaslClientAuthenticator.java:448)
        at org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.authenticate(SaslClientAuthenticator.java:262)
        at org.apache.kafka.common.network.KafkaChannel.prepare(KafkaChannel.java:173)
        at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:547)
        at org.apache.kafka.common.network.Selector.poll(Selector.java:483)
        at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:540)
        at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run(KafkaAdminClient.java:1196)
        at java.lang.Thread.run(Thread.java:748)
2020-09-27 02:08:16,019 ERROR [AdminClient clientId=adminclient-21533] Connection to node -1 (migration-kafka-topics-connect.servicebus.windows.net/xxx.xxx.xxx.xxx:9093) failed authentication due to: Invalid SASL mechanism response, server may be expecting a different protocol (org.apache.kafka.clients.NetworkClient) [kafka-admin-client-thread | adminclient-21533]

Situation

Connector state on each mirror maker.

On Mirror Maker 2 on AKS1 .

date; kubectl get KafkaMirrorMaker2 event-dt -n dev --context admin -o json | jq '.status.connectors[] | { name: .name, cnnector_state: .connector.state, task_state: .tasks[0].state }'
Sun 27 Sep 2020 12:41:41 PM JST
{
  "name": "csp-kafka->eventhub-dt.MirrorCheckpointConnector",
  "cnnector_state": "RUNNING",
  "task_state": "RUNNING"
}
{
  "name": "csp-kafka->eventhub-dt.MirrorSourceConnector",
  "cnnector_state": "RUNNING",
  "task_state": "RUNNING"
}
{
  "name": "csp-kafka->eventhub-dt.MirrorHeartbeatConnector",
  "cnnector_state": "RUNNING",
  "task_state": "RUNNING"
}

On Mirror Maker 2 on AKS2 .

date; kubectl get KafkaMirrorMaker2 event-dt -n strimzi --context ope-admin -o json | jq '.status.connectors[] | { name: .name, cnnector_state: .connector.state, task_state: .tasks[0].state }'
Sun 27 Sep 2020 12:41:59 PM JST
{
  "name": "eventhub->ea-kafka.MirrorCheckpointConnector",
  "cnnector_state": "RUNNING",
  "task_state": "UNASSIGNED"
}
{
  "name": "eventhub->ea-kafka.MirrorSourceConnector",
  "cnnector_state": "RUNNING",
  "task_state": "UNASSIGNED"
}
{
  "name": "eventhub->ea-kafka.MirrorHeartbeatConnector",
  "cnnector_state": "RUNNING",
  "task_state": "RUNNING"
}

Consuming logs directly from eventhub is okay .

Produced to Consumed from Result
Kafka cluster A Azure eventhub Success
Kafka cluster A Kafka cluster B Fail

Command used to produce .

y-watanabe@LAPTOP-IG41EBJ5:~/Development/mirrormaker2/kubernetes/ea/manifest$ kubectl -n dev --context admin run kafka-producer -ti --image=strimzi/kafka:0.17.0-kafka-2.4.0 --rm=true --restart=Never -- bin/kafka-console-producer.sh --broker-list kafka-cluster-v2-kafka-bootstrap:9092 --topic term_location_info                                                                                                                             If you don't see a command prompt, try pressing enter.                                                                                                                                                                                       >Hello World.                                                                                                                                                                                                                                >^Cpod "kafka-producer" deleted

Commands used for consuming.

Result when succeeded.

y-watanabe@LAPTOP-IG41EBJ5:~/Development/kafka_2.13-2.4.0$ bin/kafka-console-consumer.sh --bootstrap-server migration-kafka-topics-dt.servicebus.windows.net:9093 --consumer.config consumer.properties  --topic term_location_info
Hello World.
^CProcessed a total of 1 messages

Result when failed.

y-watanabe@LAPTOP-IG41EBJ5:~$ kubectl exec -it kafka-cluster-v3-kafka-0 -n strimzi --context ope-admin -- /bin/bash  bin/kafka-console-consumer.sh --bootstrap-server  kafka-cluster-v3-kafka-bootstrap:9092 --topic term_location_info
Defaulting container name to kafka.
Use 'kubectl describe pod/kafka-cluster-v3-kafka-0 -n strimzi' to see all of the containers in this pod.
OpenJDK 64-Bit Server VM warning: If the number of processors is expected to increase from one, then you should configure the number of parallel GC threads appropriately using -XX:ParallelGCThreads=N
^CProcessed a total of 0 messages
command terminated with exit code 130

Manifest for failing Mirror Maker 2.
I mostly used settings from this blog .

apiVersion: kafka.strimzi.io/v1alpha1
kind: KafkaMirrorMaker2
metadata:
  name: event-dt
spec:
  image: yuwtennis/mm2:2.4.0
  version: 2.4.0
  replicas: 1
  connectCluster: "ea-kafka"
  clusters:
  - alias: "ea-kafka"
    bootstrapServers: kafka-cluster-v3-kafka-bootstrap:9092
    config:
      config.storage.replication.factor: 1
      offset.storage.replication.factor: 1
      status.storage.replication.factor: 1
  - alias: "eventhub"
    bootstrapServers: migration-kafka-topics-dt.servicebus.windows.net:9093
    config:
      config.storage.replication.factor: 1
      offset.storage.replication.factor: 1
      status.storage.replication.factor: 1
      consumer.connections.max.idle.ms: 180000
      consumer.metadata.max.age.ms: 180000
    authentication:
      type: plain
      username: $ConnectionString
      passwordSecret:
        secretName: secret-dt
        password: eventhubspassword
    tls:
      trustedCertificates: []
  mirrors:
  - sourceCluster: "eventhub"
    targetCluster: "ea-kafka"
    sourceConnector:
      config:
        replication.factor: 1
        offset-syncs.topic.replication.factor: 1
        sync.topic.acls.enabled: "false"
        replication.policy.class: org.kafka.SimpleReplicationPolicy
    heartbeatConnector:
      config:
        heartbeats.topic.replication.factor: 1
    checkpointConnector:
      config:
        checkpoints.topic.replication.factor: 1
    topicsPattern: status_info, term_location_info, real_latest_box_info, term_no
    groupsPattern: "test01"

I use custom image to set custom replication policy.

https://hub.docker.com/layers/yuwtennis/mm2/2.4.0/images/sha256-7485fb809b205d321e26ea5250bc419a3112d18fcf0c4749241fe22d34b08e11?context=repo

Is there any setting that I am missing ?

yuwtennis commented 3 years ago

I think I am starting to figure out what is happening.

I found out that MirrrorMaker is connecting to eventhub that is not defined in passwordSecret .

2020-09-27 06:21:39,502 INFO [AdminClient clientId=adminclient-24533] Failed authentication with migration-kafka-topics-digimarc.servicebus.windows.net/13.78.106.65 (Invalid SASL mechanism response, server may be expecting a different p$otocol) (org.apache.kafka.common.network.Selector) [kafka-admin-client-thread | adminclient-24533]           
2020-09-27 06:30:58,282 INFO [AdminClient clientId=adminclient-24575] Failed authentication with migration-kafka-topics-connect.servicebus.windows.net/40.79.186.34 (Invalid SASL mechanism response, server may be expecting a different protocol) (org.apache.kafka.common.network.Selector) [kafka-admin-client-thread | adminclient-24575]

MirrorMaker2 event-dt should connect to ONLY secret event-dt .
But org.apache.kafka.clients.NetworkClient is reaching out to other SIMILAR secrets .

$ kubectl get secret -n strimzi --context ope-admin
secret-connect                                 Opaque                                1      11s
secret-digimarc                                Opaque                                1      11s
secret-dt                                      Opaque                                1      11s

In my case , secret-dt in _eventdt points to single eventhub.

apiVersion: v1
kind: Secret
metadata:
  name: secret-dt
type: Opaque
stringData:
  eventhubspassword: Endpoint=sb://migration-kafka-topics-dt.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=MYACCESSKEY

This is happening when eventhub is specified as sourceCluster . How can I define only single azure eventhub as sourceCluster when having multiple azure eventhub ?

scholzj commented 3 years ago

I assume you read the blog post https://strimzi.io/blog/2020/06/09/mirror-maker-2-eventhub/ and didn't found any differences, or? To me your config looks good. It seems to me as it finds about the other EventHubs from the one where it connects via metadata? Because I'm quite sure it would not read it from the other secrets when you don't configure them.

@ppatierno Any thoughts on this?

yuwtennis commented 3 years ago

@scholzj

Thank you for the response.

I assume you read the blog post https://strimzi.io/blog/2020/06/09/mirror-maker-2-eventhub/ and didn't found any differences, or?

Yes . I have read the blog post and it is very helpful! Comparing with the blog , I see 2 differences.

kubectl appy -f manifests/

So far I got it working by making below adjustments.

Apply MirrorMaker2 manifests one by one

e.g

kubectl apply  -n strimzi --context ope-admin -f manifest/eh-to-mm2-dt.yaml
kafkamirrormaker2.kafka.strimzi.io/event-dt created

MM2-1

date; kubectl get KafkaMirrorMaker2 event-mm2-1 -n strimzi --context ope-admin -o json | jq '.status.connectors[] | { name: .name, connector_state: .connector.state, task_state: .tasks[0].state }'
Sun 27 Sep 2020 07:40:55 PM JST
{
  "name": "eventhub-dt->ea-kafka.MirrorCheckpointConnector",
  "connector_state": "RUNNING",
  "task_state": "RUNNING"
}
{
  "name": "eventhub-dt->ea-kafka.MirrorSourceConnector",
  "connector_state": "RUNNING",
  "task_state": "RUNNING"
}
{
  "name": "eventhub-dt->ea-kafka.MirrorHeartbeatConnector",
  "connector_state": "RUNNING",
  "task_state": "RUNNING"
}

MM2-2

date; kubectl get KafkaMirrorMaker2 event-mm2-2 -n strimzi --context ope-admin -o json | jq '.status.connectors[] | { name: .name, connector_state: .connector.state, task_state: .tasks[0].state }'
Sun 27 Sep 2020 07:42:06 PM JST
{
  "name": "eventhub->ea-kafka.MirrorHeartbeatConnector",
  "connector_state": "RUNNING",
  "task_state": "RUNNING"
}
{
  "name": "eventhub->ea-kafka.MirrorSourceConnector",
  "connector_state": "RUNNING",
  "task_state": "RUNNING"
}
{
  "name": "eventhub->ea-kafka.MirrorCheckpointConnector",
  "connector_state": "RUNNING",
  "task_state": null
}

MM2-3

date; kubectl get KafkaMirrorMaker2 event-mm2-3 -n strimzi --context ope-admin -o json | jq '.status.connectors[] | { name: .name, connector_state: .connector.state, task_state: .tasks[0].state }'
Sun 27 Sep 2020 07:42:27 PM JST
{
  "name": "eventhub->ea-kafka.MirrorHeartbeatConnector",
  "connector_state": "RUNNING",
  "task_state": "RUNNING"
}
{
  "name": "eventhub->ea-kafka.MirrorCheckpointConnector",
  "connector_state": "RUNNING",
  "task_state": null
}
{
  "name": "eventhub->ea-kafka.MirrorSourceConnector",
  "connector_state": "RUNNING",
  "task_state": "RUNNING"
}

Specify jaas exclusively instead of using secret resource

e.g

apiVersion: kafka.strimzi.io/v1alpha1
kind: KafkaMirrorMaker2
metadata:
  name: event-dt
spec:
  image: yuwtennis/mm2:2.4.0
  version: 2.4.0
  replicas: 1
  connectCluster: "ea-kafka"
  clusters:
  - alias: "ea-kafka"
    bootstrapServers: kafka-cluster-v3-kafka-bootstrap:9092
    config:
      config.storage.replication.factor: 1
      offset.storage.replication.factor: 1
      status.storage.replication.factor: 1
  - alias: "eventhub-dt"
    bootstrapServers: migration-kafka-topics-dt.servicebus.windows.net:9093
    config:
      config.storage.replication.factor: 1
      offset.storage.replication.factor: 1
      status.storage.replication.factor: 1
      consumer.connections.max.idle.ms: 180000
      consumer.metadata.max.age.ms: 180000
      security.protocol: SASL_SSL
      sasl.mechanism: PLAIN
      sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="Endpoint=sb://migration-kafka-topics-dt.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey
;SharedAccessKey=MySA";
    tls:
      trustedCertificates: []
  mirrors:
  - sourceCluster: "eventhub-dt"
    targetCluster: "ea-kafka"
    sourceConnector:
      config:
        replication.factor: 1
        offset-syncs.topic.replication.factor: 1
        sync.topic.acls.enabled: "false"
        replication.policy.class: org.catseye.kafka.SimpleReplicationPolicy
    heartbeatConnector:
      config:
        heartbeats.topic.replication.factor: 1
    checkpointConnector:
      config:
        checkpoints.topic.replication.factor: 1
    topicsPattern: status_info, term_location_info, real_latest_box_info, term_no
    groupsPattern: "test"

Enable groupsPattern only in one MM2

eh-to-mm2-mm2-3.yaml:    groupsBlacklistPattern: ".*"
eh-to-mm2-mm2-2.yaml:    groupsBlacklistPattern: ".*"
eh-to-mm2-mm2-1yaml:    groupsPattern: ".*"

I still see couple of lines where MirrorMaker2 recognizes other eventhubs though.

$ kubectl logs event-dt-mirrormaker2-7597b7b96d-dl4tj  -n strimzi --context ope-admin | grep bootstrap
bootstrap.servers = [migration-kafka-topics-connect.servicebus.windows.net:9093]
bootstrap.servers = [migration-kafka-topics-digimarc.servicebus.windows.net:9093]
scholzj commented 3 years ago

@ajborley I think we discussed it somewhere already but I cannot find it. When running multiple separate MM2 instances, do the user need to change the underlying Connect topics somewhere (offset.storage.topic, config.storage.topic, status.storage.topic)? We need this for Connect, but not sure it is needed for MM2.

ajborley commented 3 years ago

Yes, if running multiple MM2 instances then there will be multiple Kafka Connect instances which will interfere with each other if they are all using the same connect topics. To set the topic names you can add the config options to the spec.clusters.config field in your KafkaMirrorMaker2 CRs for the connect clusters. E.g:

spec:
  connectCluster: "ea-kafka"
  clusters:
  - alias: "ea-kafka"
    config:
      config.storage.replication.factor: 1
      offset.storage.replication.factor: 1
      status.storage.replication.factor: 1
      config.storage.topic: event-dt_configs
      offset.storage.topic: event-dt_offsets
      status.storage.topic: event-dt_status
...
yuwtennis commented 3 years ago

@ajborley Thank you for the reply.

In my case connectCluster also has to be different among Kafka Connect Clusters. I will close this case.