version: '2'
services:
zookeeper-1:
image: confluentinc/cp-zookeeper:latest
hostname: zookeeper-1
container_name: zookeeper-1
volumes:
- /path/to/something/zk1/zk-data:/var/lib/zookeeper/data
- /path/to/something/zk1/zk-txn-logs:/var/lib/zookeeper/log
ports:
- 22181:22181
environment:
ZOOKEEPER_SERVER_ID: 1
ZOOKEEPER_CLIENT_PORT: 22181
ZOOKEEPER_TICK_TIME: 2000
ZOOKEEPER_INIT_LIMIT: 5
ZOOKEEPER_SYNC_LIMIT: 2
ZOOKEEPER_SERVERS: zookeeper-1:22888:23888;zookeeper-2:32888:33888;zookeeper-3:42888:43888
zookeeper-2:
image: confluentinc/cp-zookeeper:latest
hostname: zookeeper-2
container_name: zookeeper-2
volumes:
- /path/to/something/zk2/zk-data:/var/lib/zookeeper/data
- /path/to/something/zk2/zk-txn-logs:/var/lib/zookeeper/log
ports:
- 32181:32181
environment:
ZOOKEEPER_SERVER_ID: 2
ZOOKEEPER_CLIENT_PORT: 32181
ZOOKEEPER_TICK_TIME: 2000
ZOOKEEPER_INIT_LIMIT: 5
ZOOKEEPER_SYNC_LIMIT: 2
ZOOKEEPER_SERVERS: zookeeper-1:22888:23888;zookeeper-2:32888:33888;zookeeper-3:42888:43888
zookeeper-3:
image: confluentinc/cp-zookeeper:latest
hostname: zookeeper-3
container_name: zookeeper-3
volumes:
- /path/to/something/zk3/zk-data:/var/lib/zookeeper/data
- /path/to/something/zk3/zk-txn-logs:/var/lib/zookeeper/log
ports:
- 42181:42181
environment:
ZOOKEEPER_SERVER_ID: 3
ZOOKEEPER_CLIENT_PORT: 42181
ZOOKEEPER_TICK_TIME: 2000
ZOOKEEPER_INIT_LIMIT: 5
ZOOKEEPER_SYNC_LIMIT: 2
ZOOKEEPER_SERVERS: zookeeper-1:22888:23888;zookeeper-2:32888:33888;zookeeper-3:42888:43888
kafka-1:
image: confluentinc/cp-enterprise-kafka:latest
hostname: kafka-1
container_name: kafka-1
volumes:
- /path/to/something/kafka1/kafka-data:/var/lib/kafka/data
ports:
- 19092:19092
environment:
KAFKA_BROKER_ID: 1
KAFKA_ZOOKEEPER_CONNECT: zookeeper-1:22181,zookeeper-2:32181,zookeeper-3:42181
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://192.168.0.71:19092
KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
KAFKA_METRIC_REPORTERS: io.confluent.metrics.reporter.ConfluentMetricsReporter
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 100
CONFLUENT_METRICS_REPORTER_BOOTSTRAP_SERVERS: kafka-1:19092,kafka-2:19093,kafka-3:19094
CONFLUENT_METRICS_REPORTER_ZOOKEEPER_CONNECT: zookeeper-1:22181,zookeeper-2:32181,zookeeper-3:42181
CONFLUENT_METRICS_REPORTER_TOPIC_REPLICAS: 1
CONFLUENT_METRICS_ENABLE: 'true'
CONFLUENT_SUPPORT_CUSTOMER_ID: 'anonymous'
KAFKA_REPLICA_FETCH_MAX_BYTES: 3145728
KAFKA_MESSAGE_MAX_BYTES: 3145728
KAFKA_PRODUCER_MAX_REQUEST_SIZE: 3145728
KAFKA_CONSUMER_MAX_PARTITION_FETCH_BYTES: 3145728
depends_on:
- zookeeper-1
- zookeeper-2
- zookeeper-3
kafka-2:
image: confluentinc/cp-enterprise-kafka:latest
hostname: kafka-2
container_name: kafka-2
volumes:
- /path/to/something/kafka2/kafka-data:/var/lib/kafka/data
ports:
- 19093:19093
environment:
KAFKA_BROKER_ID: 2
KAFKA_ZOOKEEPER_CONNECT: zookeeper-1:22181,zookeeper-2:32181,zookeeper-3:42181
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://192.168.0.71:19093
KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
KAFKA_METRIC_REPORTERS: io.confluent.metrics.reporter.ConfluentMetricsReporter
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 100
CONFLUENT_METRICS_REPORTER_BOOTSTRAP_SERVERS: kafka-1:19092,kafka-2:19093,kafka-3:19094
CONFLUENT_METRICS_REPORTER_ZOOKEEPER_CONNECT: zookeeper-1:22181,zookeeper-2:32181,zookeeper-3:42181
CONFLUENT_METRICS_REPORTER_TOPIC_REPLICAS: 1
CONFLUENT_METRICS_ENABLE: 'true'
CONFLUENT_SUPPORT_CUSTOMER_ID: 'anonymous'
KAFKA_REPLICA_FETCH_MAX_BYTES: 3145728
KAFKA_MESSAGE_MAX_BYTES: 3145728
KAFKA_PRODUCER_MAX_REQUEST_SIZE: 3145728
KAFKA_CONSUMER_MAX_PARTITION_FETCH_BYTES: 3145728
depends_on:
- zookeeper-1
- zookeeper-2
- zookeeper-3
kafka-3:
image: confluentinc/cp-enterprise-kafka:latest
hostname: kafka-3
container_name: kafka-3
volumes:
- /path/to/something/kafka3/kafka-data:/var/lib/kafka/data
ports:
- 19094:19094
environment:
KAFKA_BROKER_ID: 3
KAFKA_ZOOKEEPER_CONNECT: zookeeper-1:22181,zookeeper-2:32181,zookeeper-3:42181
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://192.168.0.71:19094
KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
KAFKA_METRIC_REPORTERS: io.confluent.metrics.reporter.ConfluentMetricsReporter
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 100
CONFLUENT_METRICS_REPORTER_BOOTSTRAP_SERVERS: kafka-1:19092,kafka-2:19093,kafka-3:19094
CONFLUENT_METRICS_REPORTER_ZOOKEEPER_CONNECT: zookeeper-1:22181,zookeeper-2:32181,zookeeper-3:42181
CONFLUENT_METRICS_REPORTER_TOPIC_REPLICAS: 1
CONFLUENT_METRICS_ENABLE: 'true'
CONFLUENT_SUPPORT_CUSTOMER_ID: 'anonymous'
KAFKA_REPLICA_FETCH_MAX_BYTES: 3145728
KAFKA_MESSAGE_MAX_BYTES: 3145728
KAFKA_PRODUCER_MAX_REQUEST_SIZE: 3145728
KAFKA_CONSUMER_MAX_PARTITION_FETCH_BYTES: 3145728
depends_on:
- zookeeper-1
- zookeeper-2
- zookeeper-3
schema-registry:
image: confluentinc/cp-schema-registry:latest
hostname: schema-registry
container_name: schema-registry
ports:
- "8081:8081"
environment:
SCHEMA_REGISTRY_HOST_NAME: schema-registry
SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: zookeeper-1:22181,zookeeper-2:32181,zookeeper-3:42181
connect:
image: confluentinc/cp-kafka-connect:latest
hostname: connect
container_name: connect
depends_on:
- schema-registry
- zookeeper-1
- zookeeper-2
- zookeeper-3
- kafka-1
- kafka-2
- kafka-3
ports:
- "8083:8083"
volumes:
- /path/to/something/postgres-source-connector:/usr/share/java/postgres-source-connector
- /path/to/something/mongodb-sink-connector:/usr/share/java/mongodb-sink-connector
environment:
CONNECT_BOOTSTRAP_SERVERS: kafka-1:19092,kafka-2:19093,kafka-3:19094
CONNECT_REST_ADVERTISED_HOST_NAME: connect
CONNECT_REST_PORT: 8083
CONNECT_GROUP_ID: compose-connect-group
CONNECT_CONFIG_STORAGE_TOPIC: docker-connect-configs
CONNECT_OFFSET_STORAGE_TOPIC: docker-connect-offsets
CONNECT_STATUS_STORAGE_TOPIC: docker-connect-status
CONNECT_KEY_CONVERTER: io.confluent.connect.avro.AvroConverter
CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL: 'http://schema-registry:8081'
CONNECT_VALUE_CONVERTER: io.confluent.connect.avro.AvroConverter
CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: 'http://schema-registry:8081'
CONNECT_INTERNAL_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
CONNECT_INTERNAL_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
CONNECT_REST_ADVERTISED_HOST_NAME: "kafka-connect"
CONNECT_LOG4J_ROOT_LOGLEVEL: "INFO"
CONNECT_LOG4J_LOGGERS: "org.apache.kafka.connect.runtime.rest=WARN,org.reflections=ERROR"
CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: "1"
CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: "1"
CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: "1"
CONNECT_PLUGIN_PATH: '/usr/share/java'
CLASSPATH: /usr/share/java/monitoring-interceptors/monitoring-interceptors-5.0.0.jar
CONNECT_PRODUCER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor"
CONNECT_CONSUMER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor"
CONNECT_PRODUCER_MAX_REQUEST_SIZE: 3145728
CONNECT_CONSUMER_MAX_PARTITION_FETCH_BYTES: 3145728
control-center:
image: confluentinc/cp-enterprise-control-center:latest
hostname: control-center
container_name: control-center
depends_on:
- schema-registry
- connect
- ksql-server
- zookeeper-1
- zookeeper-2
- zookeeper-3
- kafka-1
- kafka-2
- kafka-3
ports:
- "9021:9021"
environment:
CONTROL_CENTER_BOOTSTRAP_SERVERS: kafka-1:19092,kafka-2:19093,kafka-3:19094
CONTROL_CENTER_ZOOKEEPER_CONNECT: zookeeper-1:22181,zookeeper-2:32181,zookeeper-3:42181
CONTROL_CENTER_CONNECT_CLUSTER: 'http://connect:8083'
CONTTROL_CENTER_PRODUCER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor"
CONTROL_CENTER_CONSUMER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor"
CONTROL_CENTER_KSQL_URL: "http://ksql-server:8088"
CONTROL_CENTER_CONNECT_CLUSTER: "http://connect:8083"
CONTROL_CENTER_KSQL_ADVERTISED_URL: "http://localhost:8088"
CONTROL_CENTER_SCHEMA_REGISTRY_URL: "https://schema-registry:8081"
CONTROL_CENTER_REPLICATION_FACTOR: 1
CONTROL_CENTER_INTERNAL_TOPICS_PARTITIONS: 1
CONTROL_CENTER_MONITORING_INTERCEPTOR_TOPIC_PARTITIONS: 1
CONFLUENT_METRICS_TOPIC_REPLICATION: 1
CONTROL_CENTER_CUB_KAFKA_TIMEOUT: 300
PORT: 9021
ksql-server:
image: confluentinc/cp-ksql-server:latest
hostname: ksql-server
container_name: ksql-server
depends_on:
- connect
ports:
- "8088:8088"
environment:
KSQL_CUB_KAFKA_TIMEOUT: 300
KSQL_BOOTSTRAP_SERVERS: kafka-1:19092,kafka-2:19093,kafka-3:19094
KSQL_LISTENERS: http://0.0.0.0:8088
KSQL_KSQL_SCHEMA_REGISTRY_URL: http://schema-registry:8081
KSQL_KSQL_SERVICE_ID: confluent_rmoff_01
KSQL_PRODUCER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor"
KSQL_CONSUMER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor"
KSQL_KSQL_COMMIT_INTERVAL_MS: 2000
KSQL_KSQL_CACHE_MAX_BYTES_BUFFERING: 10000000
KSQL_KSQL_AUTO_OFFSET_RESET: earliest
ksql-cli:
image: confluentinc/cp-ksql-cli:latest
hostname: ksql-cli
container_name: ksql-cli
depends_on:
- connect
- ksql-server
entrypoint: /bin/sh
tty: true
rest-proxy:
image: confluentinc/cp-kafka-rest:latest
hostname: rest-proxy
container_name: rest-proxy
depends_on:
- schema-registry
ports:
- 8082:8082
environment:
KAFKA_REST_HOST_NAME: rest-proxy
KAFKA_REST_BOOTSTRAP_SERVERS: kafka-1:19092,kafka-2:19093,kafka-3:19094
KAFKA_REST_LISTENERS: "http://0.0.0.0:8082"
KAFKA_REST_SCHEMA_REGISTRY_URL: 'http://schema-registry:8081'
KAFKA_REST_ACCESS_CONTROL_ALLOW_METHODS: 'GET,POST,PUT,DELETE,OPTIONS'
KAFKA_REST_ACCESS_CONTROL_ALLOW_ORIGIN: '*'
postgres:
image: debezium/postgres
hostname: postgres
container_name: postgres
volumes:
- /path/to/something/postgres:/var/lib/postgresql/data
environment:
POSTGRES_USER: admin
POSTGRES_PASSWORD: admin
POSTGRES_DB: some-db
ports:
- 5432:5432
我已经将Postgres连接器映射到Kafka Connect中(通过Compose中的volumes
),并且在创建新的源连接器时可以在CCC中看到它。
当我创建源连接器时,我可以看到日志消息,指示此连接器的主题已创建。我在CCC的Connect专区也看到了这个话题。我还可以看到Connect能够通过该连接器对Postgres进行身份验证。
当我对连接器中指定的表进行更改时,我看到Kafka(我有一个3的集群)正在计算谁将存储该消息。也就是说,Postgres tx日志创建了相应主题的消息以响应我的更改,因此DB、连接器和Kafka都正常工作。
我对出了什么问题不知所措。我得到的唯一迹象是最初的消息说
仔细检查是否已为向群集ControlCenter.cluster生成或使用该群集的任何客户端正确配置了监视拦截器
我的印象是,这实质上意味着我的控制中心容器配置了*_interceptor_classes
,它是我在上面粘贴的。我跟踪了这条消息上的链接,它将您带到他们的文档站点,上面写着检查提供kafka数据的web服务的响应。正如他们的文档所示,我得到的响应只有{}
,这表明Kafka说它没有数据。但确实是这样。
它是想说我也需要这些拦截器配置到连接器中吗?我不知道为任何消费者/生产者提供监视拦截器意味着什么--我还没有任何原始的Java消费者/生产者(目前为止)...目前仅限源连接器。
{
"database.server.name": "my-namespace",
"database.dbname": "my-database",
"database.hostname": "my-hostname",
"database.port": "5432",
"database.user": "admin",
"schema.whitelist": "public",
"table.whitelist": "my-database.my-table",
"connector.class": "io.debezium.connector.postgresql.PostgresConnector",
"name": "my-connector",
"database.password": "its correct"
}
当启动所有服务时,我会在相应的日志中看到以下内容,我怀疑这些日志可能会引起兴趣(下面没有特别的顺序):
control-center | 2018-09-17T20:45:02.748463792Z interceptor.classes = []
kafka-2 | 2018-09-17T20:44:56.293701931Z interceptor.classes = []
schema-registry | 2018-09-17T20:45:34.658065846Z interceptor.classes = []
connect | 2018-09-17T20:48:52.628218936Z [2018-09-17 20:48:52,628] WARN The configuration 'producer.interceptor.classes' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig)
connect | 2018-09-17T20:48:52.628472218Z [2018-09-17 20:48:52,628] WARN The configuration 'consumer.interceptor.classes' was supplied but isn't a known config. (org.apache.kafka.clients.admin.AdminClientConfig)
感谢任何帮助。谢谢!
您引用的是5.1.0
JAR,该JAR在latest
映像中不存在。如果您docker-compose exec connect bash
并转到定义的路径,您将会看到哪个版本(当前在最新
中的5.0.0
)。所以改变你作文来阅读
CLASSPATH: /usr/share/java/monitoring-interceptors/monitoring-interceptors-5.0.0.jar
查看https://github.com/rmoff/KSQL/blob/clickstream-c3/ksql-clickstream-demo/docker-compose.yml以获得使用汇流控制中心和使用Kafka Connect(如果感兴趣,还包括KSQL)的拦截器的工作Docker Compose示例。
要进一步调试,请检查:
[2018-03-02 11:39:38,594] INFO ConsumerConfig values:
[...]
interceptor.classes = [io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor]
[2018-03-02 11:39:38,806] INFO ProducerConfig values:
[...]
interceptor.classes = [io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor]
[2018-03-02 11:39:39,455] INFO creating interceptor (io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor:74)
[2018-03-02 11:39:39,456] INFO creating interceptor (io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor:70)
[2018-03-02 11:39:39,486] INFO MonitoringInterceptorConfig values:
confluent.monitoring.interceptor.publishMs = 15000
confluent.monitoring.interceptor.topic = _confluent-monitoring
(io.confluent.monitoring.clients.interceptor.MonitoringInterceptorConfig:223)
[2018-03-02 11:39:39,486] INFO MonitoringInterceptorConfig values:
confluent.monitoring.interceptor.publishMs = 15000
confluent.monitoring.interceptor.topic = _confluent-monitoring
(io.confluent.monitoring.clients.interceptor.MonitoringInterceptorConfig:223)
我已经使用docker建立了zookeeper和kafka 5.4.0版的基本三服务器集群。对于zookeeper和kafka,我指定了所有三个引导服务器。我正在尝试以一种允许一台服务器宕机且集群仍能正常运行的方式进行设置。我试图让控制中心工作,但我遇到了问题。首先,似乎只有一台服务器能够成功运行control center,而且只有当我指定它运行的引导服务器时,而不是所有三台服务器。如果我尝试在
我们正在使用confluent kafka control center,但out system health页面不起作用,总是显示重试。 我通过了文档,验证了设置是相同的,但仍然面临这个问题。 它抛出的错误是无法检查ulimited:无法运行程序ulimited错误=2没有这样的文件或目录(io.confluent.controlcenter.healthcheck.HealthCheck) 除
截止到现在,在我们所看过的程序中,总是有一系列语句从上到下精确排列,并交由 Python 忠实地执行。如果你想改变这一工作流程,应该怎么做?就像这样的情况:你需要程序作出一些决定,并依据不同的情况去完成不同的事情,例如依据每天时间的不同打印出 ‘早上好’ ‘Good Morning’ 或 ‘晚上好’ ‘Good Evening’? 正如你可能已经猜测到的那番,这是通过控制流语句来实现的。在 Pyt
Swift 提供所有多样化的控制流语句。包括 while 循环来多次执行任务; if , guard 和 switch 语句来基于特定的条件执行不同的代码分支;还有比如 break 和 continue 语句来传递执行流到你代码的另一个点上。 Swift 同样添加了 for-in 循环,它让你更简便地遍历数组、字典、范围和其他序列。 Swift 的 switch 语句同样比 C 中的对应语句多了不
本页包含内容: For 循环 While 循环 条件语句 控制转移语句(Control Transfer Statements) Swift提供了类似 C 语言的流程控制结构,包括可以多次执行任务的for和while循环,基于特定条件选择执行不同代码分支的if和switch语句,还有控制流程跳转到其他代码的break和continue语句。 除了 C 语言里面传统的 for 条件递增(for-co
我有一个关于kafka流应用程序中的控制流的基本问题。如果有两个源主题 我做了一个非常初步的测试,当记录被消费时,我偷看了一下,然后用一个简单的速溶软件打印了它们被处理的瞬间。现在 这些是主题中记录的开始和结束时间戳 主题B记录在主题A之前提取。Sysout显示主题B中的所有记录。有人能帮助理解这一点吗?我希望在编写具有多个输入源的流式应用程序时使用这种理解。 提前感谢