• [openEuler] ] 基于openEuler-20.03-LTS-SP1安装部署Kafka 2.1.0【单机版】
    一、操作系统和软件版本介绍1.操作系统为openEuler 20.03 (LTS-SP1)   可用如下命令查询:[root@localhost ~]#cat /etc/os-release [root@localhost ~]# uname -i2.软件版本   Kafka版本为:Kafka 2.1.03.JDK依赖   Kafka 是用Scala 语言开发的,运行在JVM上,因此在安装Kafka 之前需要先安装JDK。openEuler 20.03 (LTS-SP1) 默认没有安装JDK环境,需要自己自行安装。 这里安装JDK环境做简单说明,请开发者自行安装,我这里安装的是 openjdk version "1.8.0_242"。 yum install java    输入 y  查看JDK版本[root@localhost ~]#java -version二、详细安装步骤1.在根目录创建data目录   mkdmkdir data ir data2.进入data目录并下载Kafka 2.1.0 [root@localhost /]# cd data [root@localhost data]# wget https://archive.apache.org/dist/kafka/2.1.0/kafka_2.11-2.1.0.tgz3.解压并进入到解压后的目录[root@localhost data]# tar -zvxf kafka_2.11-2.1.0.tgz [root@localhost data]# cd kafka_2.11-2.1.0 4.启动zookeeper服务和kafka服务,可以在命令的结尾加个&符号,这样服务就可以在后台运行   说明:单节点的 kafka 是不用修改配置文件,直接照官网的介绍就可以,见kafka官网入门[root@localhost kafka_2.11-2.1.0]# ./bin/zookeeper-server-start.sh ./config/zookeeper.properties &[root@localhost kafka_2.11-2.1.0]#./bin/kafka-server-start.sh ./config/server.properties &三、验证1.创建一个叫"kafkademo"的topic,它只有一个分区,一个副本:[root@localhost kafka_2.11-2.1.0]# ./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic kafkademo2.发送消息,运行producer并在控制台中输一些消息,这些消息将被发送到服务端[root@localhost kafka_2.11-2.1.0]# ./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic kafkademo3.在另一个终端开启consumer,可以读取到刚才发出的消息并输出[root@localhost kafka_2.11-2.1.0]# ./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic kafkademo至此所有操作已完成,请指正。
  • [二次开发] 【Flink】flink消费kafka任务提交后报错
    【功能模块】按照官方的代码,提交flink消费kafka任务后,出现错误。在flink客户端使用的命令为bin/flink run -yt conf/ssl/ -ys 2 -m yarn-cluster -yjm 1024 -ytm 1024 -c org.mytest.stream.ReadFromKafka /opt/flink/flink.jar --topic topictest --bootstrap.servers $bs --security.protocol SASL_PLAINTEXT --sasl.kerberos.service.name kafka【截图信息】【日志信息】(可选,上传日志内容或者附件)[root@node3 flink]# bin/flink run -yt conf/ssl/ -ys 2 -m yarn-cluster -yjm 1024 -ytm 1024 -c org.mytest.stream.ReadFromKafka /opt/flink/flink.jar --topic topictest --bootstrap.servers $bs --security.protocol SASL_PLAINTEXT --sasl.kerberos.service.name kafkaSLF4J: Class path contains multiple SLF4J bindings.SLF4J: Found binding in [jar:file:/opt/hadoopclient/Flink/flink/lib/logback-classic-1.2.3.jar!/org/slf4j/impl/StaticLoggerBinder.class]SLF4J: Found binding in [jar:file:/opt/hadoopclient/HDFS/hadoop/share/hadoop/common/lib/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.SLF4J: Actual binding is of type [ch.qos.logback.classic.util.ContextSelectorStaticBinder]Cluster started: Yarn cluster with application id application_1625541283910_0063Job has been submitted with JobID f50af8c85ed9c74e813f52c71231674fjava.util.concurrent.ExecutionException: org.apache.flink.client.program.ProgramInvocationException: Job failed (JobID: f50af8c85ed9c74e813f52c71231674f)    at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)    at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1908)    at org.apache.flink.streaming.api.environment.StreamContextEnvironment.execute(StreamContextEnvironment.java:83)    at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1651)    at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1633)    at org.mytest.stream.ReadFromKafka.main(ReadFromKafka.java:21)    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)    at java.lang.reflect.Method.invoke(Method.java:498)    at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:321)    at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:205)    at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:138)    at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:700)    at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:219)    at org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:932)    at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1005)    at java.security.AccessController.doPrivileged(Native Method)    at javax.security.auth.Subject.doAs(Subject.java:422)    at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1737)    at org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41)    at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1005)Caused by: org.apache.flink.client.program.ProgramInvocationException: Job failed (JobID: f50af8c85ed9c74e813f52c71231674f)    at org.apache.flink.client.deployment.ClusterClientJobClientAdapter.lambda$null$6(ClusterClientJobClientAdapter.java:112)    at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:616)    at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:591)    at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488)    at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1975)    at org.apache.flink.client.program.rest.RestClusterClient.lambda$pollResourceAsync$21(RestClusterClient.java:565)    at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:774)    at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:750)    at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488)    at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1975)    at org.apache.flink.runtime.concurrent.FutureUtils.lambda$retryOperationWithDelay$8(FutureUtils.java:291)    at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:774)    at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:750)    at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488)    at java.util.concurrent.CompletableFuture.postFire(CompletableFuture.java:575)    at java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:943)    at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:456)    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)    at java.lang.Thread.run(Thread.java:748)Caused by: org.apache.flink.runtime.client.JobExecutionException: Job execution failed.    at org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:147)    at org.apache.flink.client.deployment.ClusterClientJobClientAdapter.lambda$null$6(ClusterClientJobClientAdapter.java:110)    ... 19 moreCaused by: org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy    at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:110)    at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:76)    at org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:192)    at org.apache.flink.runtime.scheduler.DefaultScheduler.maybeHandleTaskFailure(DefaultScheduler.java:186)    at org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:180)    at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:484)    at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:380)    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)    at java.lang.reflect.Method.invoke(Method.java:498)    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcInvocation(AkkaRpcActor.java:279)    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:194)    at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:74)    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:152)    at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26)    at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21)    at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123)    at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21)    at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170)    at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)    at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)    at akka.actor.Actor$class.aroundReceive(Actor.scala:517)    at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:225)    at akka.actor.ActorCell.receiveMessage(ActorCell.scala:592)    at akka.actor.ActorCell.invoke(ActorCell.scala:561)    at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258)    at akka.dispatch.Mailbox.run(Mailbox.scala:225)    at akka.dispatch.Mailbox.exec(Mailbox.scala:235)    at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)    at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)    at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)    at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)Caused by: java.lang.NoSuchMethodError: org.apache.flink.streaming.connectors.kafka.internal.KafkaConsumerThread.<init>(Lorg/slf4j/Logger;Lorg/apache/flink/streaming/connectors/kafka/internal/Handover;Ljava/util/Properties;Lorg/apache/flink/streaming/connectors/kafka/internals/ClosableBlockingQueue;Ljava/lang/String;JZLorg/apache/flink/metrics/MetricGroup;Lorg/apache/flink/metrics/MetricGroup;)V    at org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.<init>(KafkaFetcher.java:109)    at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer.createFetcher(FlinkKafkaConsumer.java:237)    at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:695)    at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100)    at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63)    at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:196)[root@node3 flink]# MRS版本:3.0.2flink版本:1.10.0
  • [解决方案] Spring-kafka对接华为FusionInsight Kafka 报没有方法错误
    【功能模块】开发环境集群版本:6.5.1【操作步骤&问题现象】1、参考文档:Spring-kafka对接华为FusionInsight Kafka样例2、按照参考文档操作,测试时报没有方法错误【截图信息】pom文件:运行日志:【日志信息】(可选,上传日志内容或者附件)
  • [技术干货] DRS数据同步到写KAFKA原理
    ##一、kafka简单介绍 kafka大家的第一印象是一个消息系统。但是kafka的官网的说法是: ## Apache Kafka® is a distributed streaming platform ##### kafka是一个分布式流处理平台,而流处理平台主要具备以下三种能力: 1.发布和订阅消息流,类似于消息队列或企业消息传递系统。 2.以容错的持久化方式储存消息流。 3.可以在产生消息流的时候,同时进行处理。 ##### 而kafka具备以下几个特性: 1.kafka作为一个集群可以运行在一个或者多个服务器上,这些服务器可以跨多个数据中心。 2.kafka集群存储的消息是按照topic(主题)进行分类的。 3.每个消息(也被称为记录)是由一个key,一个value和一个时间戳构成。 ##### kafka对外提供了四种核心API: 1.Producer API,允许应用程序发布消息到kafka集群上的1个或多个的topic。 2.Consumer API,允许应用程序订阅一个或多个topic,并处理这些topic的消息。 3.Streams API,允许应用程序充当一个流处理器,从1个或多个topic消费输入流,并产生一个输出流到1个或多个输出topic,有效地将输入流转换到输出流。 4.Connector API,允许构建运行可重复使用的生产者或消费者,将topic和现有的应用程序或数据系统连接起来。例如,一个关系型数据库的连接器可以捕获到该库下每一个表的变化。 ![1.png](https://bbs-img.huaweicloud.com/data/forums/attachment/forum/202107/31/095624husg0h3qjwi0o6jk.png) ##二、Producer发送数据到kafka的流程 DRS同步到kafka:DRS作为kafka的客户端,利用kafka的Producer API;将源端数据库产生的增量数据写入到目标kafka的topic上。 ##### 以mysql到kafka为例,大致流程入下: 1.将源端mysql的binlog日志记录的增量数据作为消息封装成一个Record。 2.经过拦截器,对消息进行过滤。 3.经过序列化器,将消息的key和value进行序列化,当然可以自行定义序列化规则或者自行编写序列化器。 4.消息经过分区器,确定这条消息需要发送到目标topic的分区号。如果在消息里面指定了partion字段,那么就是将消息发送到指定分区。 5.之后消息会封装从成一个一个批次汇总到RecordAccumulator。accumulator可以作为一个缓存,是kafka强大的写入性能原因之一。 6.之后会依赖一个后台唤醒的Sender线程,将数据有序的发送到leader partition所在的broker(kafka集群的每一个服务器都是一个broker)中。 7.在发送消息的过程中,kafka客户端可以从任意一个broker获取到kafka集群的metadata信息,metadata信息里面记录了kafka集群的每个topic的所有partition的信息: leader, fellow, isr, replicas等。 整体的流程如下图所示 ![2.png](https://bbs-img.huaweicloud.com/data/forums/attachment/forum/202107/31/095634ex7ofjrxinboy6fo.png) ## 三、Producer两个重要参数 1.acks决定了生产者如何在性能与数据可靠之间做取舍,官方源码中描述如下: ``` public static final String ACKS_CONFIG = "acks"; private static final String ACKS_DOC = "The number of acknowledgments the producer requires the leader to have received before considering a request complete. This controls the " + " durability of records that are sent. The following settings are allowed: " + "" + "acks=0 If set to zero then the producer will not wait for any acknowledgment from the" + " server at all. The record will be immediately added to the socket buffer and considered sent. No guarantee can be" + " made that the server has received the record in this case, and the retries configuration will not" + " take effect (as the client won't generally know of any failures). The offset given back for each record will" + " always be set to -1." + "acks=1 This will mean the leader will write the record to its local log but will respond" + " without awaiting full acknowledgement from all followers. In this case should the leader fail immediately after" + " acknowledging the record but before the followers have replicated it then the record will be lost." + "acks=all This means the leader will wait for the full set of in-sync replicas to" + " acknowledge the record. This guarantees that the record will not be lost as long as at least one in-sync replica" + " remains alive. This is the strongest available guarantee. This is equivalent to the acks=-1 setting." + ""; ``` 对于kafka来说,消息日志是按照topic分类存储的,而对于一个topic来说有partitons分区数,replication-factor副本数。 对于一个topic而言有多个分区,一个分又可以有多个副本。这些副本中,只有一个leader partition。其他都是follower partiton,仅有leader partition可以对外提供服务,follower partiton主要用于冗余备份。 而副本是存放在不同的broker上面的,因此在创建topic的时候,副本数不能大于broker的节点数的。 而acks参数呢,就是和副本有关系。 ``` acks=0:这意味着producer发送数据后,不会等待broker确认,直接发送下一条数据,性能最好 acks=1:为1意味着producer发送数据后,需要等待leader副本确认接收后,才会发送下一条数据,性能次之 acks=-1/all:这个代表的是all,意味着发送的消息写入leader partition后,等到follower从leader拉取到消息后,才会发送下一条数据,性能最差,但可靠性最强 ``` 而DRS以可靠性优先,因此设置的acks参数值为all,确保消息写入到所有可用副本后,才进行下一条写入。 2.max.in.flight.requests.per.connection,官方源码描述如下: ``` public static final String MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION = "max.in.flight.requests.per.connection"; private static final String MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_DOC = "The maximum number of unacknowledged requests the client will send on a single connection before blocking." + " Note that if this setting is set to be greater than 1 and there are failed sends, there is a risk of" + " message re-ordering due to retries (i.e., if retries are enabled)."; ``` ``` // 在InFlightRequests.java中 /** * Can we send more requests to this node? * * @param node Node in question * @return true iff we have no requests still being sent to the given node */ public boolean canSendMore(String node) { Deque queue = requests.get(node); return queue == null || queue.isEmpty() || (queue.peekFirst().send.completed() && queue.size() this.maxInFlightRequestsPerConnection); } ``` ``` // 在Sender.java中 if (guaranteeMessageOrder) { // Mute all the partitions drained for (List batchList : batches.values()) { for (ProducerBatch batch : batchList) this.accumulator.mutePartition(batch.topicPartition); } } ``` max.in.flight.requests.per.connection表示在单个连接中,最多可以忍受多少个请求处于发送中没有没有响应。kafka源码中这个参数默认是5,可以认为,在一个连接中有5个请求发送出去了,并且Producer都没有收到broker的响应。 如果这个参数大于1,由于有重试机制,可能会存在消息顺序错乱的风险。 如下图,在一个网络连接中将batch封装成不同的request,从batch队列中取出数据,按照顺序封装成不同的request(请求1... 请求5). ![3.png](https://bbs-img.huaweicloud.com/data/forums/attachment/forum/202107/31/095651pckhibcflcb5rpan.png) 如果broker在处理请求2时因为borker节点不可以等因素导致写消息到partition异常了,但是其它请求的数据都正常写入了。此时由于重试机制,Producer会将请求2重新发送。 导致broker写入到leader partition消息顺序错乱。 ![4.png](https://bbs-img.huaweicloud.com/data/forums/attachment/forum/202107/31/095703a72eewazmdued1i8.png) 而DRS为了保证数据写入到kafka是有序的,max.in.flight.requests.per.connection参数设置为1,但是这样降低了kafka的吞吐量。
  • [二次开发] 【mrs-flink产品】【调试flink-kafka程序】NoSuchMethodError
    【功能模块】mrs-flink【操作步骤&问题现象】1、使用per-job模式上传样例代码到集群上运行,执行FemaleInfoCollectionFromKafka这个样例;2、flink任务创建成功,但是任务执行时报错,显示KafkaConsumer的一个方法不存在。【截图信息】报错信息:客户端lib里已补充相关jar包,不知道具体缺少哪个?【日志信息】(可选,上传日志内容或者附件)org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy    at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:110)    at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:76)    at org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:192)    at org.apache.flink.runtime.scheduler.DefaultScheduler.maybeHandleTaskFailure(DefaultScheduler.java:186)    at org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:180)    at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:484)    at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:380)    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)    at java.lang.reflect.Method.invoke(Method.java:498)    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcInvocation(AkkaRpcActor.java:279)    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:194)    at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:74)    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:152)    at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26)    at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21)    at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123)    at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21)    at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170)    at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)    at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)    at akka.actor.Actor$class.aroundReceive(Actor.scala:517)    at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:225)    at akka.actor.ActorCell.receiveMessage(ActorCell.scala:592)    at akka.actor.ActorCell.invoke(ActorCell.scala:561)    at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258)    at akka.dispatch.Mailbox.run(Mailbox.scala:225)    at akka.dispatch.Mailbox.exec(Mailbox.scala:235)    at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)    at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)    at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)    at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)Caused by: java.lang.NoSuchMethodError: org.apache.flink.streaming.connectors.kafka.internal.KafkaConsumerThread.(Lorg/slf4j/Logger;Lorg/apache/flink/streaming/connectors/kafka/internal/Handover;Ljava/util/Properties;Lorg/apache/flink/streaming/connectors/kafka/internals/ClosableBlockingQueue;Ljava/lang/String;JZLorg/apache/flink/metrics/MetricGroup;Lorg/apache/flink/metrics/MetricGroup;)V    at org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.(KafkaFetcher.java:109)    at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer.createFetcher(FlinkKafkaConsumer.java:237)    at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:695)    at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100)    at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63)    at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:196)
  • [openEuler] 基于openEuler-20.03-LTS-SP1安装部署Kafka 2.2.0【单机版】
    一、操作系统和软件版本介绍1.操作系统为openEuler 20.03 (LTS-SP1)   可用如下命令查询:   cat /etc/os-release 2.软件版本   Kafka版本为:Kafka 2.2.0 3.JDK依赖   Kafka 是用Scala 语言开发的,运行在JVM上,因此在安装Kafka 之前需要先安装JDK。openEuler 20.03 (LTS-SP1) 默认没有安装JDK环境,需要自己自行安装。   这里安装JDK环境做简单说明,请开发者自行安装,我这里安装的是 openjdk version "1.8.0_242"。   yum install java    输入 y   查看JDK版本   java -version二、详细安装步骤1.在根目录创建data目录   mkdir data2.进入data目录并下载Kafka 2.2.0  cd data   wget https://archive.apache.org/dist/kafka/2.2.0/kafka_2.12-2.2.0.tgz3.解压并进入到解压后的目录tar -zvxf kafka_2.12-2.2.0.tgz   cd kafka_2.12-2.2.04.启动zookeeper服务和kafka服务,可以在命令的结尾加个&符号,这样服务就可以在后台运行   ./bin/zookeeper-server-start.sh ./config/zookeeper.properties &   ./bin/kafka-server-start.sh ./config/server.properties &     三、验证1.创建一个叫"euler"的topic,它只有一个分区,一个副本:   ./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic euler2.发送消息,运行producer并在控制台中输一些消息,这些消息将被发送到服务端   ./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic euler3.在另一个终端开启consumer,可以读取到刚才发出的消息并输出   ./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic euler 至此所有操作已完成,请指正。
  • [数据编排] 虚拟kafka数据源如何配置
  • [维护宝典] 如何根据strace信息定位异常连接的客户端
    问题背景描述:某局点某Kafka节点文件句柄数很高,导致此节点上Kafka的请求失败,lsof -p kafkapid发现Kafka进程有大量socket泄漏,如下图所示:问题分析步骤:1. 执行ls -l /proc/kafkapid/fd >> fd.txt命令查看Kafka进程的文件描述符2. 执行trace -T -tt -f -F output.txt -p kafkapid命令打出kafka进程的调用信息3. 几分钟后再次执行执行ls -l /proc/kafkapid/fd >> fd1.txt命令查看Kafka进程的文件描述符4. 对比1和3步骤中的文件描述符,查找增加的socket连接,多查找几个增加socket连接,根据socket对应的id号从2步骤中获取的信息中查看对应的id连接调用信息5. 如果查找socket连接对应的客户端,可以通过accept查找,如下图所示此查询查询到对应的socketfd对应的所有的客户端连接,多次查找后可对比客户端连接来进一步确认客户端的范围。6. 通过accept查找的原因是socket通信中accept函数获取客户端的连接:int accept(int sockfd,struct sockaddr * addr,socklen_t * addrlen);sockfd的参数为listen()函数返回的监听套接字;addr是一个传出参数,表示客户端的地址,该参数设置为NULL时,表示不关心客户端的地址。addrlen为一个传入传出参数,传入时为函数调用时提供参数addr的长度,传出时为客户端地址结构体的实际长度;accept的返回值也是一个套接字,该套接字用于与本次通信的客户端进行数据交互。
  • [问题求助] 线下kafka如何对接mrs的kafka?
    大数据的项目。想把客户线下kafka的消息传递到mrs的kafka中。这个可以如何实现?
  • [维护宝典] Kafka使用 kafka-consumer-groups.sh --list查询不到消费者组
    问题描述:FusionInsight C80版本,通过Flink提交消费Kafka的任务,消费Kafka的消费者组信息通过kafka-consumer-groups.sh --list命令查询不到,但是通过kafka-consumer-groups.sh --describe可以查询到信息。问题分析:1、C80版本kafka-consumer-groups.sh命令使用list查询groupid源码分析如下:a. kafka-consumer-groups.sh命令的入口是执行ConsumerGroupCommand中的main函数b. ConsumerGroupCommand中的main函数执行时根据使用的是--zookeeper还是--bootstrap-server判断consumerGroupService是ZkConsumerGroupService还是KafkaConsumerGroupService。如果命令中有--list,就执行listGroups().foreach(println(_))c. KafkaConsumerGroupService中listGroups()的实现是通过adminClient获取所有的consumerGroup,源码如下:d. AdminClient中获取listAllConsumerGroupsFlattened的实现是获取所有的group组并过滤出所有协议是consumer的group组,源码如下:2. 提交一个flink作业,groupid设置为“testgroup”,使用Kafka中的AdminClient来获取listAllGroupsFlattened的信息,执行结果如下:如上图所示:执行结果中GroupOverview中testgroup对应的协议为空,而不是consumer,所以,使用list查询不到结果。3. Flink作业消费kafka时,GroupOverview中groupid对应的协议为空的原因是:flink不是直接调用的kafka consumer client端的消费接口,而是通过自己的逻辑去消费、只用kafka中的__consumer_offsets保存数据。4. 此问题在651版本已经解决,651版本中增加的协议为空的判断逻辑,源码如下所示:
  • [维护宝典] FusionInsight Kafka创建Topic失败原因及解决方案
    一、 无权限创建Topic会报错“ERROR org.apache.zookeeper.KeeperException$NoAuthException: KeeperErrorCode = NoAuth for /config/topics/topicName。解决办法是使用具体kafkaadmin权限的用户。二、 Quota限制创建Topic会报错“Nodes number exceed quota limit for /config/topics/topicName”。解决办法是:检查Kafka全部配置中Quota配置quota.number是否是250000,如果不是改为此值。同步zk配置。FusionInsight Manager页面“集群->Zookeeper->更多->同步配置”,同步配置操作每5分钟执行一次,最多等5分钟,zk客户端执行listquota /kafka查看quota的count值是否为250000。zk客户端执行操作方法如下:进入FI客户端(例如/opt/hadoopclient)执行命令source bigdata_env执行命令kinit 用户名(普通模式跳过)执行sh -server zk业务IP:24002三、 副本个数大于正常节点数创建Topic会报错“ERROR org.apache.kafka.common.errors.InvalidReplicationFactorException: Replication factor: X larger than available brokers: X”解决办法是:解决异常节点问题或减少副本数。四、 客户端版本低客户端是低版本而服务端是高版本时,创建Topic可能不会报错但是查看topic的describe信息时leader为none。原因是zk上报没有权限访问“/brokers/topics/topicName”目录,此目录下无法创建partitions目录。解决办法是:使用与服务端对应的客户端版本。
  • [问题求助] DMS Kafka能力上能匹配Confluent 5.2.2版本吗?
    华为云DMS Kafka能否在能力上匹配下Confluent
  • [二次开发] 【flink产品】【flink功能】flink跨集群访问开启Kerberos认证的kafka
    【功能模块】FI集群的flink 是开启的Kerberos认证 再flink-conf.yml中配置过的。现在需要用flink消费另外一个集群中开启Kerberos认证的kafka  拿到了kafka集群的的认证文件 krb5.conf jaas.conf user.keytab。【操作步骤&问题现象】1、整个flink应用是需要自己集群的认证 ,怎么在一个flink程序中认证另外一个kafka ,将其当作该Flink应用的source源,然后处理呢?拜托各位大佬提供一下处理思路!
  • [问题求助] DLI Flink作业如何动态感知Kafka分区的变化
    我使用Flink Opensource SQL, 采用Flink 1.10版本。初期Flink作业规划的Kafka的分区数partition设置过小或过大,后期需要更改Kafka区分数,很麻烦。如何不停止作业实现自动感知?
  • [其他] FI651 filebeat对接安全模式kafka
    再使用kafka的过程中,用到filebeat采集日志到kfka,非安全模式的kafka开源直接使用,但是安全模式的kafka,尝试使用类似logstash对接安全模式kafka,未能连通,请问filebeat对接安全的kafka需要如何配置?
总条数:186 到第
上滑加载中