报错java.lang.ClassNotFoundException: io.netty.channel.ChannelHandler类找不到 about flink-connector-redis

[jeff-zou]

暂时不支持source，只支持left join, 后面几天我会把source支持加上。

[WhiteLie98]

您的意思是目前不支持只从redis中get数据然后写到其他数据库吗？

还有个疑惑，目前我用redis获取数据并与kafka的数据碰撞，出现了java.lang.ClassNotFoundException报错，即这个类找不到org.apache.flink.calcite.shaded.com.google.common.cache.Cache。
无论是直接用您1.3.1的release版本jar，还是进行去test编译后的依赖包，都有这个提示……

现有flink环境的lib文件夹：

具体脱敏后代码如下：

CREATE DATABASE IF NOT EXISTS test_source;
CREATE DATABASE IF NOT EXISTS test_sink;

CREATE TABLE test_source.test_redis (
  redis_key string,
  redis_value string
) WITH (
  'connector' = 'redis',
  'password' = 'XXXXX',
  'host' = 'XXXXX',
  'port' = 'XXXXX',
  'redis-mode' = 'single',
  'command'='get',
  'maxIdle'='2',
  'minIdle'='1',
  'lookup.cache.max-rows'='10',
  'lookup.cache.ttl'='10',
  'lookup.max-retries'='3'
);

CREATE TABLE test_source.test_kafka (
  kafka_key varchar(50),
  kafka_value varchar(50),
  pt AS PROCTIME()
) WITH (
  'connector' = 'kafka',
  'topic' = 'topicXXXX',
  'properties.bootstrap.servers' = 'XXXXX:XXXXX',
  'properties.group.id' = 'XXXX',
  'scan.startup.mode' = 'latest-offset',
  'format' = 'json',
  'json.ignore-parse-errors'='true',
  'properties.security.protocol' = 'SASL_PLAINTEXT',
  'properties.sasl.mechanism' = 'SCRAM-SHA-256',
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.scram.ScramLoginModule required username="XXXX" password="XXXX";'
  );

CREATE TABLE IF NOT EXISTS test_sink.test_sr_kafka2redis (
  sr_key varchar(50),
  sr_value varchar(50),
  PRIMARY KEY (sr_key) NOT ENFORCED
) WITH (
  'connector' = 'starrocks',
  'jdbc-url'='jdbc:mysql://XXXXX:XXXXX?characterEncoding=utf-8&useSSL=false&serverTimezone=Asia/Shanghai',
  'load-url'='XXXXX:XXXXX',
  'sink.properties.strip_outer_array' = 'true',
  'sink.properties.format' = 'json',
  'database-name' = 'XXXXX',
  'username' = 'XXXXX',
  'password' = 'XXXXX',
  'table-name' = 'test_sr_kafka2redis',
  'sink.buffer-flush.interval-ms' = '60000',
  'sink.parallelism' = '1' 
);

INSERT INTO test_sink.test_sr_kafka2redis
select
  k.kafka_key as sr_key, 
  k.kafka_value as sr_value
from test_source.test_kafka k
left join test_source.test_redis FOR SYSTEM_TIME AS OF k.pt r 
on r.redis_key = concat('XXXXX::', k.kafka_key);

[jeff-zou]

你的项目没引入这个包：
<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-planner_${scala.binary.version}</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency>

[WhiteLie98]

你的项目没引入这个包： <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-planner_${scala.binary.version}</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency>

谢谢！这个问题确实解决了之前报的错误，但因为之前是有flink-table-planner-loader，这两个包不能同时在lib文件里……

不过，现在又出现了一个找不到的类问题，如图：

我加入了netty-transport-native-epoll包，但我看编译的时候还有其他的netty包，是都要放进lib环境下吗？