登录
当前位置： 首页 > 知识库问答 >
问题：

库伯涅茨笔下的Kafka——标志着集体协调者的死亡

方昊阳
2023-03-14

我对库伯内特斯很陌生，想用它来设置Kafka和动物园管理员。我能够使用有状态集在库伯内特斯中设置阿帕奇Kafka和动物园管理员。我按照这个和这个来构建我的清单文件。我制作了1个Kafka和动物园管理员的复制品，并且还使用了持久卷。所有 Pod 都在运行并准备就绪。

我试图公开kafka，并通过指定nodePort（30010）来使用Service。似乎这会将kafka暴露给外部世界，在那里他们可以向kafka代理发送消息并从中消费。

但是在我的Java应用程序中，我创建了一个消费者，并将bootstrapServer添加为。

INFO o.a.k.c.c.i.AbstractCoordinator - Discovered coordinator kafka-0.kafka-hs.default.svc.cluster.local:9093 (id: 2147483647 rack: null) for group workerListener.
INFO o.a.k.c.c.i.AbstractCoordinator - Marking the coordinator kafka-0.kafka-hs.default.svc.cluster.local:9093 (id: 2147483647 rack: null) dead for group workerListener

有趣的是，当我使用 kubectl 命令测试集群时，我能够生成和使用消息：

kubectl run -ti --image=gcr.io/google_containers/kubernetes-kafka:1.0-10.2.1 produce --restart=Never --rm \
 -- kafka-console-producer.sh --topic test --broker-list kafka-0.kafka-hs.default.svc.cluster.local:9093 done;

kubectl run -ti --image=gcr.io/google_containers/kubernetes-kafka:1.0-10.2.1 consume --restart=Never --rm -- kafka-console-consumer.sh --topic test --bootstrap-server kafka-0.kafka-hs.default.svc.cluster.local:9093

有人能给我指出正确的方向，为什么它将协调员标记为死亡？

kafka.yml

apiVersion: v1
kind: Service
metadata:
  name: kafka-hs
  labels:
    app: kafka
spec:
  ports:
  - port: 9093
    name: server
  clusterIP: None
  selector:
    app: kafka
---
apiVersion: v1
kind: Service
metadata:
  name: kafka-cs
  labels:
    app: kafka
spec:
  type: NodePort
  ports:
  - port: 9093
    nodePort: 30010
    protocol: TCP
  selector:
    app: kafka
---
apiVersion: apps/v1beta1
kind: StatefulSet
metadata:
  name: kafka
spec:
  serviceName: kafka-hs
  replicas: 1
  podManagementPolicy: Parallel
  updateStrategy:
    type: RollingUpdate
  template:
    metadata:
      labels:
        app: kafka
    spec:
      affinity:
        podAntiAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            - labelSelector:
                matchExpressions:
                  - key: "app"
                    operator: In
                    values:
                    - kafka
              topologyKey: "kubernetes.io/hostname"
        podAffinity:
          preferredDuringSchedulingIgnoredDuringExecution:
             - weight: 1
               podAffinityTerm:
                 labelSelector:
                    matchExpressions:
                      - key: "app"
                        operator: In
                        values:
                        - zk
                 topologyKey: "kubernetes.io/hostname"
      terminationGracePeriodSeconds: 300
      containers:
      - name: k8skafka
        imagePullPolicy: Always
        image: gcr.io/google_containers/kubernetes-kafka:1.0-10.2.1
        resources:
          requests:
            memory: "1Gi"
            cpu: "0.5"
        ports:
        - containerPort: 9093
          name: server
        command:
        - sh
        - -c
        - "exec kafka-server-start.sh /opt/kafka/config/server.properties --override broker.id=${HOSTNAME##*-} \
          --override listeners=PLAINTEXT://:9093 \
          --override zookeeper.connect=zk-cs.default.svc.cluster.local:2181 \
          --override log.dir=/var/lib/kafka \
          --override auto.create.topics.enable=true \
          --override auto.leader.rebalance.enable=true \
          --override background.threads=10 \
          --override compression.type=producer \
          --override delete.topic.enable=false \
          --override leader.imbalance.check.interval.seconds=300 \
          --override leader.imbalance.per.broker.percentage=10 \
          --override log.flush.interval.messages=9223372036854775807 \
          --override log.flush.offset.checkpoint.interval.ms=60000 \
          --override log.flush.scheduler.interval.ms=9223372036854775807 \
          --override log.retention.bytes=-1 \
          --override log.retention.hours=168 \
          --override log.roll.hours=168 \
          --override log.roll.jitter.hours=0 \
          --override log.segment.bytes=1073741824 \
          --override log.segment.delete.delay.ms=60000 \
          --override message.max.bytes=1000012 \
          --override min.insync.replicas=1 \
          --override num.io.threads=8 \
          --override num.network.threads=3 \
          --override num.recovery.threads.per.data.dir=1 \
          --override num.replica.fetchers=1 \
          --override offset.metadata.max.bytes=4096 \
          --override offsets.commit.required.acks=-1 \
          --override offsets.commit.timeout.ms=5000 \
          --override offsets.load.buffer.size=5242880 \
          --override offsets.retention.check.interval.ms=600000 \
          --override offsets.retention.minutes=1440 \
          --override offsets.topic.compression.codec=0 \
          --override offsets.topic.num.partitions=50 \
          --override offsets.topic.replication.factor=3 \
          --override offsets.topic.segment.bytes=104857600 \
          --override queued.max.requests=500 \
          --override quota.consumer.default=9223372036854775807 \
          --override quota.producer.default=9223372036854775807 \
          --override replica.fetch.min.bytes=1 \
          --override replica.fetch.wait.max.ms=500 \
          --override replica.high.watermark.checkpoint.interval.ms=5000 \
          --override replica.lag.time.max.ms=10000 \
          --override replica.socket.receive.buffer.bytes=65536 \
          --override replica.socket.timeout.ms=30000 \
          --override request.timeout.ms=30000 \
          --override socket.receive.buffer.bytes=102400 \
          --override socket.request.max.bytes=104857600 \
          --override socket.send.buffer.bytes=102400 \
          --override unclean.leader.election.enable=true \
          --override zookeeper.session.timeout.ms=6000 \
          --override zookeeper.set.acl=false \
          --override broker.id.generation.enable=true \
          --override connections.max.idle.ms=600000 \
          --override controlled.shutdown.enable=true \
          --override controlled.shutdown.max.retries=3 \
          --override controlled.shutdown.retry.backoff.ms=5000 \
          --override controller.socket.timeout.ms=30000 \
          --override default.replication.factor=1 \
          --override fetch.purgatory.purge.interval.requests=1000 \
          --override group.max.session.timeout.ms=300000 \
          --override group.min.session.timeout.ms=6000 \
          --override inter.broker.protocol.version=0.10.2-IV0 \
          --override log.cleaner.backoff.ms=15000 \
          --override log.cleaner.dedupe.buffer.size=134217728 \
          --override log.cleaner.delete.retention.ms=86400000 \
          --override log.cleaner.enable=true \
          --override log.cleaner.io.buffer.load.factor=0.9 \
          --override log.cleaner.io.buffer.size=524288 \
          --override log.cleaner.io.max.bytes.per.second=1.7976931348623157E308 \
          --override log.cleaner.min.cleanable.ratio=0.5 \
          --override log.cleaner.min.compaction.lag.ms=0 \
          --override log.cleaner.threads=1 \
          --override log.cleanup.policy=delete \
          --override log.index.interval.bytes=4096 \
          --override log.index.size.max.bytes=10485760 \
          --override log.message.timestamp.difference.max.ms=9223372036854775807 \
          --override log.message.timestamp.type=CreateTime \
          --override log.preallocate=false \
          --override log.retention.check.interval.ms=300000 \
          --override max.connections.per.ip=2147483647 \
          --override num.partitions=1 \
          --override producer.purgatory.purge.interval.requests=1000 \
          --override replica.fetch.backoff.ms=1000 \
          --override replica.fetch.max.bytes=1048576 \
          --override replica.fetch.response.max.bytes=10485760 \
          --override reserved.broker.max.id=1000 "
        env:
        - name: KAFKA_HEAP_OPTS
          value : "-Xmx512M -Xms512M"
        - name: KAFKA_OPTS
          value: "-Dlogging.level=INFO"
        volumeMounts:
        - name: kafka-pv-volume
          mountPath: /var/lib/kafka
        readinessProbe:
          exec:
           command:
            - sh
            - -c
            - "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server=localhost:9093"
      securityContext:
        runAsUser: 0
        fsGroup: 1000
  volumeClaimTemplates:
  - metadata:
      name: kafka-pv-volume
    spec:
      storageClassName: manual
      accessModes: [ "ReadWriteOnce" ]
      resources:
        requests:
          storage: 1Gi

动物园管理员

apiVersion: v1
kind: Service
metadata:
  name: zk-hs
  labels:
    app: zk
spec:
  ports:
  - port: 2888
    name: server
  - port: 3888
    name: leader-election
  clusterIP: None
  selector:
    app: zk
---
apiVersion: v1
kind: Service
metadata:
  name: zk-cs
  labels:
    app: zk
spec:
  ports:
  - port: 2181
    name: client
  selector:
    app: zk
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: zk
spec:
  selector:
    matchLabels:
      app: zk
  serviceName: zk-hs
  replicas: 1
  updateStrategy:
    type: RollingUpdate
  podManagementPolicy: Parallel
  template:
    metadata:
      labels:
        app: zk
    spec:
      affinity:
        podAntiAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            - labelSelector:
                matchExpressions:
                  - key: "app"
                    operator: In
                    values:
                    - zk
              topologyKey: "kubernetes.io/hostname"
      containers:
      - name: kubernetes-zookeeper
        imagePullPolicy: Always
        image: "k8s.gcr.io/kubernetes-zookeeper:1.0-3.4.10"
        resources:
          requests:
            memory: "1Gi"
            cpu: "0.5"
        ports:
        - containerPort: 2181
          name: client
        - containerPort: 2888
          name: server
        - containerPort: 3888
          name: leader-election
        command:
        - sh
        - -c
        - "start-zookeeper \
          --servers=1 \
          --data_dir=/var/lib/zookeeper/data \
          --data_log_dir=/var/lib/zookeeper/data/log \
          --conf_dir=/opt/zookeeper/conf \
          --client_port=2181 \
          --election_port=3888 \
          --server_port=2888 \
          --tick_time=2000 \
          --init_limit=10 \
          --sync_limit=5 \
          --heap=512M \
          --max_client_cnxns=60 \
          --snap_retain_count=3 \
          --purge_interval=12 \
          --max_session_timeout=40000 \
          --min_session_timeout=4000 \
          --log_level=INFO"
        readinessProbe:
          exec:
            command:
            - sh
            - -c
            - "zookeeper-ready 2181"
          initialDelaySeconds: 10
          timeoutSeconds: 5
        livenessProbe:
          exec:
            command:
            - sh
            - -c
            - "zookeeper-ready 2181"
          initialDelaySeconds: 10
          timeoutSeconds: 5
        volumeMounts:
        - name: pv-volume
          mountPath: /var/lib/zookeeper
      securityContext:
        runAsUser: 0
        fsGroup: 1000
  volumeClaimTemplates:
  - metadata:
      name: pv-volume
    spec:
      storageClassName: manual
      accessModes: [ "ReadWriteOnce" ]
      resources:
        requests:
          storage: 1Gi

编辑：

我将日志级别更改为 TRACE。这些是我得到的日志

2018-01-11 18:56:24,617 TRACE o.a.k.c.NetworkClient - Completed receive from node -1, for key 3, received {brokers=[{node_id=0,host=kafka-0.kafka-hs.default.svc.cluster.local,port=9093,rack=null}],cluster_id=LwSLmJpTQf6tSKPsfvriIg,controller_id=0,topic_metadata=[{topic_error_code=0,topic=mdm.worker.request,is_internal=false,partition_metadata=[{partition_error_code=0,partition_id=0,leader=0,replicas=[0],isr=[0]}]}]}
2018-01-11 18:56:24,621 DEBUG o.a.k.c.Metadata - Updated cluster metadata version 2 to Cluster(id = LwSLmJpTQf6tSKPsfvriIg, nodes = [kafka-0.kafka-hs.default.svc.cluster.local:9093 (id: 0 rack: null)], partitions = [Partition(topic = mdm.worker.request, partition = 0, leader = 0, replicas = [0], isr = [0])])
2018-01-11 18:56:24,622 TRACE o.a.k.c.NetworkClient - Completed receive from node -1, for key 10, received {error_code=0,coordinator={node_id=0,host=kafka-0.kafka-hs.default.svc.cluster.local,port=9093}}
2018-01-11 18:56:24,624 DEBUG o.a.k.c.c.i.AbstractCoordinator - Received GroupCoordinator response ClientResponse(receivedTimeMs=1515678984622, latencyMs=798, disconnected=false, requestHeader={api_key=10,api_version=0,correlation_id=0,client_id=consumer-1}, responseBody=FindCoordinatorResponse(throttleTimeMs=0, errorMessage='null', error=NONE, node=kafka-0.kafka-hs.default.svc.cluster.local:9093 (id: 0 rack: null))) for group workerListener
2018-01-11 18:56:24,625 INFO o.a.k.c.c.i.AbstractCoordinator - Discovered coordinator kafka-0.kafka-hs.default.svc.cluster.local:9093 (id: 2147483647 rack: null) for group workerListener.
2018-01-11 18:56:24,625 DEBUG o.a.k.c.NetworkClient - Initiating connection to node 2147483647 at kafka-0.kafka-hs.default.svc.cluster.local:9093.
2018-01-11 18:56:24,633 DEBUG o.a.k.c.NetworkClient - Error connecting to node 2147483647 at kafka-0.kafka-hs.default.svc.cluster.local:9093:
java.io.IOException: Can't resolve address: kafka-0.kafka-hs.default.svc.cluster.local:9093
    at org.apache.kafka.common.network.Selector.connect(Selector.java:195)
    at org.apache.kafka.clients.NetworkClient.initiateConnect(NetworkClient.java:762)
    at org.apache.kafka.clients.NetworkClient.ready(NetworkClient.java:224)
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.tryConnect(ConsumerNetworkClient.java:462)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$GroupCoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:598)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$GroupCoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:579)
    at org.apache.kafka.clients.consumer.internals.RequestFuture$1.onSuccess(RequestFuture.java:204)
    at org.apache.kafka.clients.consumer.internals.RequestFuture.fireSuccess(RequestFuture.java:167)
    at org.apache.kafka.clients.consumer.internals.RequestFuture.complete(RequestFuture.java:127)
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler.fireCompletion(ConsumerNetworkClient.java:488)
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.firePendingCompletedRequests(ConsumerNetworkClient.java:348)
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:262)
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:208)
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:184)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:214)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:200)
    at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:286)
    at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1078)
    at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1043)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:614)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.lang.Thread.run(Thread.java:748)
Caused by: java.nio.channels.UnresolvedAddressException: null
    at sun.nio.ch.Net.checkAddress(Net.java:101)
    at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:622)
    at org.apache.kafka.common.network.Selector.connect(Selector.java:192)
    ... 22 common frames omitted
2018-01-11 18:56:24,634 INFO o.a.k.c.c.i.AbstractCoordinator - Marking the coordinator kafka-0.kafka-hs.default.svc.cluster.local:9093 (id: 2147483647 rack: null) dead for group workerListener
2018-01-11 18:56:24,735 TRACE o.a.k.c.NetworkClient - Found least loaded node kafka-0.kafka-hs.default.svc.cluster.local:9093 (id: 0 rack: null)
2018-01-11 18:56:24,735 DEBUG o.a.k.c.c.i.AbstractCoordinator - Sending GroupCoordinator request for group workerListener to broker kafka-0.kafka-hs.default.svc.cluster.local:9093 (id: 0 rack: null)
2018-01-11 18:56:24,735 DEBUG o.a.k.c.NetworkClient - Initiating connection to node 0 at kafka-0.kafka-hs.default.svc.cluster.local:9093.
2018-01-11 18:56:24,736 DEBUG o.a.k.c.NetworkClient - Error connecting to node 0 at kafka-0.kafka-hs.default.svc.cluster.local:9093:
java.io.IOException: Can't resolve address: kafka-0.kafka-hs.default.svc.cluster.local:9093
    at org.apache.kafka.common.network.Selector.connect(Selector.java:195)
    at org.apache.kafka.clients.NetworkClient.initiateConnect(NetworkClient.java:762)
    at org.apache.kafka.clients.NetworkClient.ready(NetworkClient.java:224)
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.trySend(ConsumerNetworkClient.java:408)
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:223)
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:208)
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:184)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:214)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:200)
    at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:286)
    at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1078)
    at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1043)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:614)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.lang.Thread.run(Thread.java:748)
Caused by: java.nio.channels.UnresolvedAddressException: null
    at sun.nio.ch.Net.checkAddress(Net.java:101)
    at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:622)
    at org.apache.kafka.common.network.Selector.connect(Selector.java:192)
    ... 15 common frames omitted
2018-01-11 18:56:24,737 TRACE o.a.k.c.NetworkClient - Removing node kafka-0.kafka-hs.default.svc.cluster.local:9093 (id: 0 rack: null) from least loaded node selection: is-blacked-out: true, in-flight-requests: 0
2018-01-11 18:56:24,737 TRACE o.a.k.c.NetworkClient - Least loaded node selection failed to find an available node
2018-01-11 18:56:24,738 DEBUG o.a.k.c.NetworkClient - Give up sending metadata request since no node is available

共有1个答案

洪飞鸿
2023-03-14

上周我和你有同样的问题，并且解决了，所以有可能在库伯内特斯外面揭露Kafka！

解决方案：在您的Kafkabroker配置中。yaml您应该将群集外部IP映射到本地DNS

kafka-I.kafka-hs.default.svc.cluster.local:9093

如何：

将它们添加到您的服务器属性文件中：

listener.security.protocol.map=INTERNAL_PLAINTEXT:PLAINTEXT,EXTERNAL_PLAINTEXT:PLAINTEXT
inter.broker.listener.name=INTERNAL_PLAINTEXT

如果你有一个在服务器属性之前运行的初始化，你应该添加这些：

# add unique label to each pod
kubectl label pods ${HOSTNAME} kafka-set-component=${HOSTNAME}

EXTERNAL_LISTENER_IP=<YOUR_KUBERNETES_CLUSTER_EXTERNAL_IP>
EXTERNAL_LISTENER_PORT=$((30093 + ${HOSTNAME##*-}))

sed -i "s/#listeners=PLAINTEXT:\/\/:9092/listeners=INTERNAL_PLAINTEXT:\/\/0.0.0.0:9092,EXTERNAL_PLAINTEXT:\/\/0.0.0.0:9093/" /etc/kafka/server.properties
sed -i "s/#advertised.listeners=PLAINTEXT:\/\/your.host.name:9092/advertised.listeners=INTERNAL_PLAINTEXT:\/\/$HOSTNAME.broker.kafka.svc.cluster.local:9092,EXTERNAL_PLAINTEXT:\/\/$EXTERNAL_LISTENER_IP:$EXTERNAL_LISTENER_PORT/" /etc/kafka/server.properties

否则，您应该找到在服务器中添加替换配置的方法。属性。

请注意，您必须在server.properties文件中对这些行进行注释

#listeners=PLAINTEXT://:9092
#advertised.listeners=PLAINTEXT://your.host.name:9092

服务：创建无外设服务以映射本地 DNS 和您拥有的每个代理的服务：

# A headless service to create DNS records
---
apiVersion: v1
kind: Service
metadata:
  name: broker
  namespace: kafka
spec:
  ports:
  - port: 9092
  # [podname].broker.kafka.svc.cluster.local
  clusterIP: None
  selector:
    app: kafka
---
apiVersion: v1
kind: Service
metadata:
  name: broker-0
  namespace: kafka
spec:
  type: NodePort
  ports:
  - port: 9093
    nodePort: 30093
  selector:
    kafka-set-component: kafka-0
---
apiVersion: v1
kind: Service
metadata:
  name: broker-1
  namespace: kafka
spec:
  type: NodePort
  ports:
  - port: 9093
    nodePort: 30094
  selector:
    kafka-set-component: kafka-1
---
apiVersion: v1
kind: Service
metadata:
  name: broker-2
  namespace: kafka
spec:
  type: NodePort
  ports:
  - port: 9093
    nodePort: 30095
  selector:
    kafka-set-component: kafka-2

注意：-如果您在GKE上运行：

  1. YOUR_KUBERNETES_CLUSTER_EXTERNAL_IP在服务器属性中声明，可以通过g云计算实例列表找到
  2. 此外，您必须授予防火墙 g 云计算防火墙的权限 - 规则创建Kafka - 外部 - 允许tcp：30093，tcp：30094，tcp：30095
 类似资料：
  • Kafka笔下的消费者

    我正在与Kafka和阿帕奇·Flink合作。我正在尝试使用apache Flink中的一个kafka主题中的记录（这些记录是avro格式的）。下面是我正在尝试的一段代码。 使用自定义反序列化器来反序列化主题中的avro记录。 我发送到主题“test-topic”的数据的Avro模式如下所示。 我正在使用的自定义反序列化器如下所示。 我的flink应用程序就是这样写的。 我得到的输出是{“name”

  • 面向ConcurrentKafkaListenerContainerFactory的Spring-Kafka消费者群体协调

    关于斯普林斯-Kafka在某些场景中的行为，我有几个问题。任何答案或指针都将是伟大的。 背景：我正在构建一个与外部API对话并返回确认的kafka消费者。我的配置如下所示： 让我们来讨论一个场景，其中调用了消费者的kafkalistener，它调用但是服务关闭，然后根据，它重试3次然后失败，然后调用errorhandler，从而停止KafkalistenerEndpointRegistry。这将关

  • 库伯内特斯没有音调集群

    我正在尝试让cadence在kubernetes集群上运行。然而，我注意到Cadence服务器初始化中有一个bug，它阻止Cassandra脚本正确初始化模式。https://github.com/uber/cadence/issues/1713:所以我想我会手动完成这一步。我执行了以下步骤- < li >在docker compose上从https://raw . githubuserconte

  • 跨消费者的Kafka日志分布

    apache kafka文档提到以下内容： 如果所有使用者实例具有相同的使用者组，那么记录将有效地在使用者实例上进行负载平衡。 如果所有的使用者实例都有不同的使用者组，那么每个记录都将广播给所有的使用者进程。

  • Kafka划分到Kafka消费者/消费者群体的映射

    我正在阅读Kafka常见问题解答，他们如下所示。 •每个分区不会被每个使用者组中的多个使用者线程/进程使用。这允许每个进程以单线程方式使用，以保证分区内的使用者的顺序（如果我们将有序消息分割成一个分区并将它们传递给多个使用者，即使这些消息是按顺序存储的，它们有时也会被无序地处理）。 有没有可能，

  • 新的 Kafka Consumer（消费者）集成

    这部分包含新的 Apache Kafka consumer API. 兼容性 Apache Kafka 版本 0.10+ 写入Kafka 您可以通过创建 org.apache.storm.kafka.bolt.KafkaBolt 实例并将其作为组件附加到您的topology.如果您使用 trident ,您可以通过使用以下对象完成 org.apache.storm.kafka.trident.Tr

相关问答

Kafka划分到Kafka消费者/消费者群体的映射Kafka消费者团体标识共享问题如果Zookeeper集群中的节点死亡，Kafka集群不可用Kafka关于消费者群体的问题Kafka流库会杀死空闲的流线程吗？

相关文章

PHP：利用exit（）; 或者死（）; 在标头之后（“ Location：”）；如何更改标签标题框的大小和ttk笔记本标签的字体？请你说一下http协议会话结束标志怎么截出来？Kafka-经纪人：小组协调员不可用page.evaluate内部的伪造者日志

相关阅读

4399后端笔试体验OPPO多媒体c++笔试笔试体验四合一 策划岗（笔经）DefaultMQPushConsumer消费者重试消息和死信消息C++标准库中的字符串

相关工具

不死鸟Google地图JSP标签库语法着色Highlight死鱼微信类快速集成下拉上拉刷新

相关文档

开发学习笔记合集周志华《机器学习》学习笔记Jims 的学习笔记Go 语言标准库C 标准库 中文版
快捷导航： 新手教程 算法原理 架构设计 Java进阶 数据库进阶 大厂专栏 面试经验 编程笔记 编程问答 所有专题 文档资料 工具软件 电子书籍 小牛导航
在线工具： 房贷计算器 个税计算器 Linux命令查询 Json格式化 正则表达式 颜色转换 AES加解密 SHA1加密 MD5加密 毒鸡汤 字数统计 随机密码生成 进制转换 Base64编解码 励志句子
Copyright © 2019-2024 小牛知识库@xnip.cn. All Rights Reserved.