flink测试redis sink报错

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
Redis 开源版,标准版 2GB
推荐场景:
搭建游戏排行榜
云数据库 Tair(兼容Redis),内存型 2GB
简介: 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
相关文章
|
1月前
|
消息中间件 分布式计算 大数据
大数据-123 - Flink 并行度 相关概念 全局、作业、算子、Slot并行度 Flink并行度设置与测试
大数据-123 - Flink 并行度 相关概念 全局、作业、算子、Slot并行度 Flink并行度设置与测试
105 0
|
1月前
|
Java 流计算
Flink-03 Flink Java 3分钟上手 Stream 给 Flink-02 DataStreamSource Socket写一个测试的工具!
Flink-03 Flink Java 3分钟上手 Stream 给 Flink-02 DataStreamSource Socket写一个测试的工具!
37 1
Flink-03 Flink Java 3分钟上手 Stream 给 Flink-02 DataStreamSource Socket写一个测试的工具!
|
21天前
|
NoSQL 测试技术 Redis
Redis 性能测试
10月更文挑战第21天
22 2
|
1月前
|
存储 NoSQL 大数据
大数据-51 Redis 高可用方案CAP-AP 主从复制 一主一从 全量和增量同步 哨兵模式 docker-compose测试
大数据-51 Redis 高可用方案CAP-AP 主从复制 一主一从 全量和增量同步 哨兵模式 docker-compose测试
33 3
|
1月前
|
存储 Prometheus NoSQL
大数据-44 Redis 慢查询日志 监视器 慢查询测试学习
大数据-44 Redis 慢查询日志 监视器 慢查询测试学习
24 3
|
1月前
|
缓存 NoSQL Ubuntu
大数据-39 Redis 高并发分布式缓存 Ubuntu源码编译安装 云服务器 启动并测试 redis-server redis-cli
大数据-39 Redis 高并发分布式缓存 Ubuntu源码编译安装 云服务器 启动并测试 redis-server redis-cli
55 3
|
1月前
|
消息中间件 NoSQL Kafka
Flink-10 Flink Java 3分钟上手 Docker容器化部署 JobManager TaskManager Kafka Redis Dockerfile docker-compose
Flink-10 Flink Java 3分钟上手 Docker容器化部署 JobManager TaskManager Kafka Redis Dockerfile docker-compose
38 4
|
1月前
|
分布式计算 Java 大数据
大数据-122 - Flink Time Watermark Java代码测试实现Tumbling Window
大数据-122 - Flink Time Watermark Java代码测试实现Tumbling Window
31 0
|
1月前
|
消息中间件 NoSQL Kafka
大数据-116 - Flink DataStream Sink 原理、概念、常见Sink类型 配置与使用 附带案例1:消费Kafka写到Redis
大数据-116 - Flink DataStream Sink 原理、概念、常见Sink类型 配置与使用 附带案例1:消费Kafka写到Redis
129 0
|
1月前
|
消息中间件 NoSQL Kafka
Flink-05 Flink Java 3分钟上手 Redis FlinkJedisPoolConfig 从Kafka写入Redis FlinkKafkaConsumer消费 结果写入Redis
Flink-05 Flink Java 3分钟上手 Redis FlinkJedisPoolConfig 从Kafka写入Redis FlinkKafkaConsumer消费 结果写入Redis
33 0