java - Avro 协议(protocol)中的 Flume SSL 问题

Question

我在将事件从一项服务传递到另一项服务时遇到问题。当我禁用 SSL 时，我的事件在另一端正确接收。但是当我启用 SSL 时，出现异常。

org.apache.flume.EventDeliveryException: NettyAvroRpcClient { host: 192.168.168.197, port: 41422 }: Failed to send event
    at org.apache.flume.api.NettyAvroRpcClient.append(NettyAvroRpcClient.java:250)
    at com.ashok.sink.Main.main(Main.java:22)
Caused by: org.apache.flume.EventDeliveryException: NettyAvroRpcClient { host: 192.168.168.197, port: 41422 }: Exception thrown from remote handler
    at org.apache.flume.api.NettyAvroRpcClient.waitForStatusOK(NettyAvroRpcClient.java:397)
    at org.apache.flume.api.NettyAvroRpcClient.append(NettyAvroRpcClient.java:297)
    at org.apache.flume.api.NettyAvroRpcClient.append(NettyAvroRpcClient.java:238)
    ... 1 more
Caused by: java.util.concurrent.ExecutionException: java.io.IOException: NettyTransceiver closed
    at org.apache.avro.ipc.CallFuture.get(CallFuture.java:128)
    at org.apache.flume.api.NettyAvroRpcClient.waitForStatusOK(NettyAvroRpcClient.java:389)
    ... 3 more
Caused by: java.io.IOException: NettyTransceiver closed
    at org.apache.avro.ipc.NettyTransceiver.disconnect(NettyTransceiver.java:338)
    at org.apache.avro.ipc.NettyTransceiver.access$200(NettyTransceiver.java:59)
    at org.apache.avro.ipc.NettyTransceiver$NettyClientAvroHandler.handleUpstream(NettyTransceiver.java:496)
    at org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:558)
    at org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:786)
    at org.jboss.netty.handler.codec.frame.FrameDecoder.cleanup(FrameDecoder.java:489)
    at org.jboss.netty.handler.codec.frame.FrameDecoder.channelClosed(FrameDecoder.java:372)
    at org.jboss.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:88)
    at org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:558)
    at org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:553)
    at org.jboss.netty.channel.Channels.fireChannelClosed(Channels.java:476)
    at org.jboss.netty.channel.socket.nio.AbstractNioWorker.close(AbstractNioWorker.java:736)
    at org.jboss.netty.channel.socket.nio.NioWorker.read(NioWorker.java:89)
    at org.jboss.netty.channel.socket.nio.AbstractNioWorker.processSelectedKeys(AbstractNioWorker.java:471)
    at org.jboss.netty.channel.socket.nio.AbstractNioWorker.run(AbstractNioWorker.java:332)
    at org.jboss.netty.channel.socket.nio.NioWorker.run(NioWorker.java:35)
    at org.jboss.netty.util.ThreadRenamingRunnable.run(ThreadRenamingRunnable.java:102)
    at org.jboss.netty.util.internal.DeadLockProofWorker$1.run(DeadLockProofWorker.java:42)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
    at java.lang.Thread.run(Thread.java:744)

我正在使用 apache flume 1.6(最新)

我的配置是:

agent.sources.ashok-source.type = avro
agent.sources.ashok-source.channels = ashok-channel
agent.sources.ashok-source.bind = 192.168.168.197
agent.sources.ashok-source.port = 41422
agent.sources.ashok-source.ssl = true
agent.sources.ashok-source.keystore = S:/Ashok/keystore/ashokKeystore.jks
agent.sources.ashok-source.keystore-password = ashok-123
agent.sources.ashok-source.keystore-type = jks
agent.sinks.ashok-sink.type = com.ashok.sink.AshokSink
agent.sinks.ashok-sink.channel = ashok-channel
agent.sinks.ashok-sink.ssl = true
agent.sinks.ashok-sink.truststore = S:/Ashok/keystore/ashokKeystore.jks
agent.sinks.ashok-sink.truststore-password = ashok-123
agent.sinks.ashok-sink.truststore-type = JKS
agent.channels.ashok-channel.type = file
agent.channels.ashok-channel.checkpointDir = ../channel/ashok/checkpoint
agent.channels.ashok-channel.dataDirs = ../channel/ashok/data

我的代码:

RpcClient client = null;
try {
        client = RpcClientFactory.getDefaultInstance("192.168.168.197", 41422);
        Map<String, String> header = new HashMap<String, String>();
        header.put("head", "er");
        Event event = EventBuilder.withBody(ByteArrayUtil.getBytes("Data"), header);
        client.append(event);
    } catch (Exception e) {
        e.printStackTrace();
    }

我使用以下链接创建了我的 JKS 文件。 https://www.sslshopper.com/article-how-to-create-a-self-signed-certificate-using-java-keytool.html

我犯了一些错误，谁能帮帮我。

Answer 1

我没有使用 RpcClientFactory.getDefaultInstance("192.168.168.197", 41422);，而是在运行时使用带有信任库属性的 ​​RpcClientFactory.getInstance(properies) 修复了我的问题问题。

Map<String, String> header = new HashMap<String, String>();
TransTO transTO = new TransTO(1, "Ashok", true);
header.put("print", "yes");
Event event = EventBuilder.withBody(ByteArrayUtil.getBytes(transTO), header);
Properties prop = new Properties();
prop.setProperty("hosts.h1", "192.168.2.62:41422");
prop.setProperty("hosts", "h1");
prop.setProperty("compression-type", "deflate");
prop.setProperty("truststore-type", "JKS");
prop.setProperty("ssl", "true");
prop.setProperty("truststore", "D:/New folder/cacerts.jks");
prop.setProperty("truststore-password", "ashok-123");
RpcClient client = RpcClientFactory.getInstance(prop);
client.append(event);