flink测试redis sink报错

本文涉及的产品
Redis 开源版,标准版 2GB
推荐场景:
搭建游戏排行榜
云数据库 Tair(兼容Redis),内存型 2GB
实时计算 Flink 版,5000CU*H 3个月
简介: flink测试redis sink报错

具体报错

1    [Source: Custom Source -> Sink: Unnamed (1/1)#0] ERROR org.apache.flink.streaming.connectors.redis.RedisSink  - Redis has not been properly initialized: 
redis.clients.jedis.exceptions.JedisConnectionException: Could not get a resource from the pool
  at redis.clients.util.Pool.getResource(Pool.java:50)
  at redis.clients.jedis.JedisPool.getResource(JedisPool.java:99)
  at org.apache.flink.streaming.connectors.redis.common.container.RedisContainer.getInstance(RedisContainer.java:250)
  at org.apache.flink.streaming.connectors.redis.common.container.RedisContainer.open(RedisContainer.java:85)
  at org.apache.flink.streaming.connectors.redis.RedisSink.open(RedisSink.java:174)
  at org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:34)
  at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:102)
  at org.apache.flink.streaming.api.operators.StreamSink.open(StreamSink.java:46)
  at org.apache.flink.streaming.runtime.tasks.OperatorChain.initializeStateAndOpenOperators(OperatorChain.java:437)
  at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:574)
  at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.call(StreamTaskActionExecutor.java:100)
  at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:554)
  at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:756)
  at org.apache.flink.runtime.taskmanager.Task.run(Task.java:563)
  at java.lang.Thread.run(Thread.java:750)
Caused by: redis.clients.jedis.exceptions.JedisConnectionException: java.net.ConnectException: Connection refused: connect
  at redis.clients.jedis.Connection.connect(Connection.java:164)
  at redis.clients.jedis.BinaryClient.connect(BinaryClient.java:80)
  at redis.clients.jedis.BinaryJedis.connect(BinaryJedis.java:1676)
  at redis.clients.jedis.JedisFactory.makeObject(JedisFactory.java:87)
  at org.apache.commons.pool2.impl.GenericObjectPool.create(GenericObjectPool.java:861)
  at org.apache.commons.pool2.impl.GenericObjectPool.borrowObject(GenericObjectPool.java:435)
  at org.apache.commons.pool2.impl.GenericObjectPool.borrowObject(GenericObjectPool.java:363)
  at redis.clients.util.Pool.getResource(Pool.java:48)
  ... 14 more
Caused by: java.net.ConnectException: Connection refused: connect
  at java.net.DualStackPlainSocketImpl.waitForConnect(Native Method)
  at java.net.DualStackPlainSocketImpl.socketConnect(DualStackPlainSocketImpl.java:85)
  at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:350)
  at java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:206)
  at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:188)
  at java.net.PlainSocketImpl.connect(PlainSocketImpl.java:172)
  at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:392)
  at java.net.Socket.connect(Socket.java:607)
  at redis.clients.jedis.Connection.connect(Connection.java:158)
  ... 21 more
Exception in thread "main" org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
  at org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:144)
  at org.apache.flink.runtime.minicluster.MiniClusterJobClient.lambda$getJobExecutionResult$3(MiniClusterJobClient.java:137)
  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.runtime.rpc.akka.AkkaInvocationHandler.lambda$invokeRpc$0(AkkaInvocationHandler.java:237)
  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$1.onComplete(FutureUtils.java:1081)
  at akka.dispatch.OnComplete.internal(Future.scala:264)
  at akka.dispatch.OnComplete.internal(Future.scala:261)
  at akka.dispatch.japi$CallbackBridge.apply(Future.scala:191)
  at akka.dispatch.japi$CallbackBridge.apply(Future.scala:188)
  at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:60)
  at org.apache.flink.runtime.concurrent.Executors$DirectExecutionContext.execute(Executors.java:73)
  at scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:68)
  at scala.concurrent.impl.Promise$DefaultPromise.$anonfun$tryComplete$1(Promise.scala:284)
  at scala.concurrent.impl.Promise$DefaultPromise.$anonfun$tryComplete$1$adapted(Promise.scala:284)
  at scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:284)
  at akka.pattern.PromiseActorRef.$bang(AskSupport.scala:573)
  at akka.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:22)
  at akka.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:21)
  at scala.concurrent.Future.$anonfun$andThen$1(Future.scala:532)
  at scala.concurrent.impl.Promise.liftedTree1$1(Promise.scala:29)
  at scala.concurrent.impl.Promise.$anonfun$transform$1(Promise.scala:29)
  at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:60)
  at akka.dispatch.BatchingExecutor$AbstractBatch.processBatch(BatchingExecutor.scala:55)
  at akka.dispatch.BatchingExecutor$BlockableBatch.$anonfun$run$1(BatchingExecutor.scala:91)
  at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:12)
  at scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:81)
  at akka.dispatch.BatchingExecutor$BlockableBatch.run(BatchingExecutor.scala:91)
  at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:40)
  at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(ForkJoinExecutorConfigurator.scala:44)
  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: org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy
  at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:138)
  at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:82)
  at org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:207)
  at org.apache.flink.runtime.scheduler.DefaultScheduler.maybeHandleTaskFailure(DefaultScheduler.java:197)
  at org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:188)
  at org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:677)
  at org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:79)
  at org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:435)
  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:305)
  at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:212)
  at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:77)
  at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:158)
  at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26)
  at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21)
  at scala.PartialFunction.applyOrElse(PartialFunction.scala:123)
  at scala.PartialFunction.applyOrElse$(PartialFunction.scala:122)
  at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21)
  at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
  at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172)
  at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172)
  at akka.actor.Actor.aroundReceive(Actor.scala:517)
  at akka.actor.Actor.aroundReceive$(Actor.scala:515)
  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)
  ... 4 more
Caused by: redis.clients.jedis.exceptions.JedisConnectionException: Could not get a resource from the pool
  at redis.clients.util.Pool.getResource(Pool.java:50)
  at redis.clients.jedis.JedisPool.getResource(JedisPool.java:99)
  at org.apache.flink.streaming.connectors.redis.common.container.RedisContainer.getInstance(RedisContainer.java:250)
  at org.apache.flink.streaming.connectors.redis.common.container.RedisContainer.open(RedisContainer.java:85)
  at org.apache.flink.streaming.connectors.redis.RedisSink.open(RedisSink.java:174)
  at org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:34)
  at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:102)
  at org.apache.flink.streaming.api.operators.StreamSink.open(StreamSink.java:46)
  at org.apache.flink.streaming.runtime.tasks.OperatorChain.initializeStateAndOpenOperators(OperatorChain.java:437)
  at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:574)
  at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.call(StreamTaskActionExecutor.java:100)
  at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:554)
  at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:756)
  at org.apache.flink.runtime.taskmanager.Task.run(Task.java:563)
  at java.lang.Thread.run(Thread.java:750)
Caused by: redis.clients.jedis.exceptions.JedisConnectionException: java.net.ConnectException: Connection refused: connect
  at redis.clients.jedis.Connection.connect(Connection.java:164)
  at redis.clients.jedis.BinaryClient.connect(BinaryClient.java:80)
  at redis.clients.jedis.BinaryJedis.connect(BinaryJedis.java:1676)
  at redis.clients.jedis.JedisFactory.makeObject(JedisFactory.java:87)
  at org.apache.commons.pool2.impl.GenericObjectPool.create(GenericObjectPool.java:861)
  at org.apache.commons.pool2.impl.GenericObjectPool.borrowObject(GenericObjectPool.java:435)
  at org.apache.commons.pool2.impl.GenericObjectPool.borrowObject(GenericObjectPool.java:363)
  at redis.clients.util.Pool.getResource(Pool.java:48)
  ... 14 more
Caused by: java.net.ConnectException: Connection refused: connect
  at java.net.DualStackPlainSocketImpl.waitForConnect(Native Method)
  at java.net.DualStackPlainSocketImpl.socketConnect(DualStackPlainSocketImpl.java:85)
  at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:350)
  at java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:206)
  at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:188)
  at java.net.PlainSocketImpl.connect(PlainSocketImpl.java:172)
  at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:392)
  at java.net.Socket.connect(Socket.java:607)
  at redis.clients.jedis.Connection.connect(Connection.java:158)
  ... 21 more

根据报错定位到一个错误栈:Caused by: redis.clients.jedis.exceptions.JedisConnectionException: java.net.ConnectException: Connection refused: connect


应该是redis连接不上,确认配置(host、password、port、timeout等)没有问题之后检查逐步排查其他问题


1、先检查了redis状态

[root@master src]# ps -aux | grep redis
root      11455  0.0  0.0 162396  7796 ?        Ssl  20:42   0:01 redis-server 127.0.0.1:6379
root      58166  0.0  0.0 112828   988 pts/0    S+   21:19   0:00 grep --color=auto redis
[root@master src]# netstat -nltp | grep redis
tcp        0      0 127.0.0.1:6379          0.0.0.0:*               LISTEN      11455/redis-server
[root@master src]# redis-cli
127.0.0.1:6379>

redis正常状态且能正常使用

2、检查防火墙是否关闭或者是否放行redis端口(redis默认端口6379)

[root@master src]# systemctl status firewalld
● firewalld.service - firewalld - dynamic firewall daemon
   Loaded: loaded (/usr/lib/systemd/system/firewalld.service; disabled; vendor preset: enabled)
   Active: inactive (dead)
     Docs: man:firewalld(1)
[root@master src]#

防火墙为关闭状态

3、查看redis配置里的保护模式是否关闭(redis3.2版本以上有这个属性)

查看redis配置里的bind是否关闭

[root@master redis-6.0.8]# cat redis.conf | grep protect
# Protected mode is a layer of security protection, in order to avoid that
# When protected mode is on and if:
# By default protected mode is enabled. You should disable it only if
protected-mode yes
# If the master is password protected (using the "requirepass" configuration
# on the internet. It's just a protection layer against misuse of the instance.
# So use the 'requirepass' option to protect your instance.
[root@master redis-6.0.8] # cat redis.conf |grep bind
# By default, if no "bind" configuration directive is specified, Redis listens
# the "bind" configuration directive, followed by one or more IP addresses.
# bind 192.168.1.100 10.0.0.1
# bind 127.0.0.1 ::1
# internet, binding to all the interfaces is dangerous and will expose the
# following bind directive, that will force Redis to listen only into
bind 127.0.0.1
# 1) The server is not binding explicitly to a set of addresses using the
#    "bind" directive.
# are explicitly listed using the "bind" directive.

可以看到redis保护模式合bind 参数是开启的

  • 将配置中的yes改为no
  • 将bind哪一行注释掉
  • 之后重新启动即可解决问题
相关实践学习
基于Redis实现在线游戏积分排行榜
本场景将介绍如何基于Redis数据库实现在线游戏中的游戏玩家积分排行榜功能。
云数据库 Redis 版使用教程
云数据库Redis版是兼容Redis协议标准的、提供持久化的内存数据库服务,基于高可靠双机热备架构及可无缝扩展的集群架构,满足高读写性能场景及容量需弹性变配的业务需求。 产品详情:https://www.aliyun.com/product/kvstore     ------------------------------------------------------------------------- 阿里云数据库体验:数据库上云实战 开发者云会免费提供一台带自建MySQL的源数据库 ECS 实例和一台目标数据库 RDS实例。跟着指引,您可以一步步实现将ECS自建数据库迁移到目标数据库RDS。 点击下方链接,领取免费ECS&RDS资源,30分钟完成数据库上云实战!https://developer.aliyun.com/adc/scenario/51eefbd1894e42f6bb9acacadd3f9121?spm=a2c6h.13788135.J_3257954370.9.4ba85f24utseFl
相关文章
|
22天前
|
消息中间件 分布式计算 大数据
大数据-123 - Flink 并行度 相关概念 全局、作业、算子、Slot并行度 Flink并行度设置与测试
大数据-123 - Flink 并行度 相关概念 全局、作业、算子、Slot并行度 Flink并行度设置与测试
74 0
|
23小时前
|
存储 SQL 关系型数据库
2024Mysql And Redis基础与进阶操作系列(1)作者——LJS[含MySQL的下载、安装、配置详解步骤及报错对应解决方法]
Mysql And Redis基础与进阶操作系列(1)之[MySQL的下载、安装、配置详解步骤及报错对应解决方法]
|
23小时前
|
SQL 关系型数据库 MySQL
|
21小时前
|
NoSQL 安全 关系型数据库
2024Mysql And Redis基础与进阶操作系列(6)作者——LJS[含MySQL 多表之一对一/多;多对多;多表联合查询等详解步骤及常见报错问题所对应的解决方法]
MySQL 多表之一对一/多;多对多;多表联合之交叉连接;内连接;左、右、外、满、连接;子查询及关键字;自连接查询等详解步骤及常见报错问题所对应的解决方法
|
22小时前
|
SQL NoSQL 关系型数据库
2024Mysql And Redis基础与进阶操作系列(5)作者——LJS[含MySQL DQL基本查询:select;简单、排序、分组、聚合、分组、分页等详解步骤及常见报错问题所对应的解决方法]
MySQL DQL基本查询:select;简单、排序、分组、聚合、分组、分页、INSERT INTO SELECT / FROM查询结合精例等详解步骤及常见报错问题所对应的解决方法
|
22小时前
|
SQL NoSQL 关系型数据库
|
23天前
|
Java 流计算
Flink-03 Flink Java 3分钟上手 Stream 给 Flink-02 DataStreamSource Socket写一个测试的工具!
Flink-03 Flink Java 3分钟上手 Stream 给 Flink-02 DataStreamSource Socket写一个测试的工具!
34 1
Flink-03 Flink Java 3分钟上手 Stream 给 Flink-02 DataStreamSource Socket写一个测试的工具!
|
22小时前
|
NoSQL 关系型数据库 MySQL
2024Mysql And Redis基础与进阶操作系列(4-2)作者——LJS[含MySQL非空、唯一性、PRIMARY KEY、自增列/自增约束举例说明等详解步骤及常见报错问题对应的解决方法]
24MySQL非空、唯一性、PRIMARY KEY、自增列/自增约束举例说明等详解步骤及常见报错问题对应的解决方法(4-2) 学不会你来砍我!!!
|
22天前
|
分布式计算 Java 大数据
大数据-122 - Flink Time Watermark Java代码测试实现Tumbling Window
大数据-122 - Flink Time Watermark Java代码测试实现Tumbling Window
28 0
|
3月前
|
运维 Java Devops
阿里云云效操作报错合集之在流水线增加单元测试报错,是什么导致的
本合集将整理呈现用户在使用过程中遇到的报错及其对应的解决办法,包括但不限于账户权限设置错误、项目配置不正确、代码提交冲突、构建任务执行失败、测试环境异常、需求流转阻塞等问题。阿里云云效是一站式企业级研发协同和DevOps平台,为企业提供从需求规划、开发、测试、发布到运维、运营的全流程端到端服务和工具支撑,致力于提升企业的研发效能和创新能力。