我试图使用Kerberos从ActiveMQ-Camel桥向Kafka (Cloudera)发送消息。
骆驼.xml剪刀是:
<log message="Started The Producer Route" />
<to uri="kafka://10.100.70.00:9092?topic=MyEvents.s1.v1&brokers=10.100.70.00:9092&requestTimeoutMs=305000&retries=3&keySerializerClass=org.apache.kafka.common.serialization.ByteArraySerializer&saslMechanism=GSSAPI&serializerClass=org.apache.kafka.common.serialization.ByteArraySerializer&securityProtocol=PLAINTEXT&saslKerberosServiceName=kafka"/>
这是日志中的kafka客户端配置:
acks = 1
batch.size = 16384
bootstrap.servers = [10.148.70.74:9092]
buffer.memory = 33554432
client.id =
compression.type = none
connections.max.idle.ms = 540000
enable.idempotence = false
interceptor.classes = []
key.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer
linger.ms = 0
max.block.ms = 60000
max.in.flight.requests.per.connection = 5
max.request.size = 1048576
metadata.max.age.ms = 300000
metric.reporters = []
metrics.num.samples = 2
metrics.recording.level = INFO
metrics.sample.window.ms = 30000
partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
receive.buffer.bytes = 65536
reconnect.backoff.max.ms = 1000
reconnect.backoff.ms = 50
request.timeout.ms = 305000
retries = 3
retry.backoff.ms = 100
sasl.jaas.config = null
sasl.kerberos.kinit.cmd = /usr/bin/kinit
sasl.kerberos.min.time.before.relogin = 60000
sasl.kerberos.service.name = kafka
sasl.kerberos.ticket.renew.jitter = 0.05
sasl.kerberos.ticket.renew.window.factor = 0.8
sasl.mechanism = GSSAPI
security.protocol = PLAINTEXT (**SASL_PLAINTEXT not supported**)
send.buffer.bytes = 131072
ssl.cipher.suites = null
ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
ssl.endpoint.identification.algorithm = null
ssl.key.password = null
ssl.keymanager.algorithm = SunX509
ssl.keystore.location = null
ssl.keystore.password = null
ssl.keystore.type = JKS
ssl.protocol = TLS
ssl.provider = null
ssl.secure.random.implementation = null
ssl.trustmanager.algorithm = PKIX
ssl.truststore.location = null
ssl.truststore.password = null
ssl.truststore.type = JKS
transaction.timeout.ms = 60000
transactional.id = null
value.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer
日志级别:调试
Jaas文件:
KafkaClient {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
storeKey=true
keyTab="./user.keytab"
useTicketCache=false
serviceName="kafka"
principal=" Group/user@DOMAIN.LAN";
};
出口:
KAFKA_OPTS="-Djava.security.auth.login.config=/opt/activemq/conf/Jaas.conf"
当我发送消息时,我在调试级别收到以下日志,但消息未被发送:
java.io.EOFException
at org.apache.kafka.common.network.NetworkReceive.readFromReadableChannel(NetworkReceive.java:124)[kafka-clients-1.1.0.jar:]
at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:93)[kafka-clients-1.1.0.jar:]
at org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:235)[kafka-clients-1.1.0.jar:]
at org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:196)[kafka-clients-1.1.0.jar:]
at org.apache.kafka.common.network.Selector.attemptRead(Selector.java:557)[kafka-clients-1.1.0.jar:]
at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:495)[kafka-clients-1.1.0.jar:]
at org.apache.kafka.common.network.Selector.poll(Selector.java:424)[kafka-clients-1.1.0.jar:]
at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:460)[kafka-clients-1.1.0.jar:]
at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:239)[kafka-clients-1.1.0.jar:]
at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:163)[kafka-clients-1.1.0.jar:]
at java.lang.Thread.run(Thread.java:748)[:1.8.0_171
在INFO级别,我只在日志中看到这个:
WARN | [Producer clientId=producer-1] Bootstrap broker 10.100.70.00:9092 (id: -1 rack: null) disconnected | org.apache.kafka.clients.NetworkClient | kafka-producer-network-thread | producer-1
为什么我会收到此错误?请帮忙!
这个错误是由未经授权的用户向Kafka发送消息造成的。
可以通过验证密钥表文件作为先决条件来缓解此类问题:
不应有任何错误。
我正在尝试从目录中选取一个文件,拆分一个文件,并将每一个拆分行添加到ActiveMQ中。我在这个过程中遇到了异常处理的问题。假设目录中的文件是一个二进制文件(可执行文件),那么splitter会显示org.apache.camel.runtimeCamelException和java.nio.charset.MalFormedInputException异常。如果出现这种情况,那么我需要捕获这些异
在camel spring ws中,有没有办法将自定义主体(例如自定义bean)和http状态返回为500? 我试过了 然后在processor公共类ExceptionProcessor实现processor{ } 但是即使身体是我想要的,超文本传输协议状态总是200。 你能帮我吗? 更多信息:我正在使用camel 2.20.2版
无法弄清楚这是怎么回事-试图设置一个路由,只看到cxf连接到soap web服务(我不关心实际数据,也不期望数据实际“工作”,但它不断引发一个我不理解的异常: 我想知道我是否正确配置了它。我想这可能是一个丢失的jar,但当我尝试引入其他jar时,导致了依赖冲突 我正在使用maven依赖项“camel-cxf”来加载我所有的jar配置 “原因:org.apache.cxf.bus.extension
我正在将CSV文件输入集成到一个更大的系统中,我想对其进行设置,以便将CSV中未正确解析的各行(由)发送到医院队列,并将所有其他工作行转换为一个对象并聚合到一个列表中。但是,我很难将
我们在Camel中定义了一个路由,并且必须找出是否在处理器中抛出了异常。当我们只有一个处理器时,Camel会在sendbody()方法中重新抛出异常。如果有前面的拆分/聚合,则不会抛出异常。因此下面示例的结果是 如果我省略所有从。拆分到。完成大小(1)输出是 任何想法如何找出,如果一个异常发生后分裂? 为了在之后检查异常,我们找到了一个解决方案。我们向onException()注册了一个Error
Camel Version 2.22.0 运行时:SpringBoot:2.0.2.RELEASE JDK版本:1.8。0_121 弹性公网IP:收件人列表。 问题:从接收方列表的并行进程引发的异常未在路由级别的onExc的子句中捕获。 下面是DSL