我做了kafka java生产者.
但控制台说错误. kafka服务器在aws上.和制作人在我的Mac上.
然而kara服务器是可以访问的.当我从制作人发送消息时,kafka服务器显示“已接受的连接…”.
有什么问题?
1 [main] INFO kafka.utils.VerifiableProperties - Verifying properties
28 [main] INFO kafka.utils.VerifiableProperties - Property      Metadata.broker.list is overridden to xxxxxx:9092
28 [main] INFO kafka.utils.VerifiableProperties - Property serializer.class is overridden to kafka.serializer.StringEncoder
137 [main] INFO kafka.client.ClientUtils$- Fetching Metadata from broker id:0,host: xxxxxx,port:9092 with correlation id 0 for 1 topic(s) Set(words_topic)
189 [main] ERROR kafka.producer.SyncProducer - Producer connection to xxxxxx:9092 unsuccessful
198 [main] WARN kafka.client.ClientUtils$- Fetching topic Metadata with correlation id 0 for topics [Set(words_topic)] from broker [id:0,port:9092] Failed
199 [main] ERROR kafka.utils.Utils$- fetching topic Metadata for topics [Set(words_topic)] from broker [ArrayBuffer(id:0,port:9092)] Failed

它是kafka控制台

[2015-01-27 05:23:33,767] DEBUG Accepted connection from /xxxxx on /xxxx:9092. sendBufferSize [actual|requested]: [212992|1048576] recvBufferSize [actual|requested]: [212992|1048576] (kafka.network.Acceptor)
[2015-01-27 05:23:33,767] DEBUG Processor 1 listening to new connection from /xxxx:65307 (kafka.network.Processor)
[2015-01-27 05:23:33,872] INFO Closing socket connection to /xxxx. (kafka.network.Processor)
[2015-01-27 05:23:33,873] DEBUG Closing connection from /xxxx:65307 (kafka.network.Processor)

这是我的代码.

Properties props = new Properties();
    props.put("Metadata.broker.list","?????:9092");
    props.put("serializer.class","kafka.serializer.StringEncoder");
    ProducerConfig config = new ProducerConfig(props);
    Producer<String,String> producer = new Producer<String,String>(config);

    // Now we break each word from the paragraph
        for (String word :
            MetaMORPHOSIS_opening_Para.split("\\s")) {
        // Create message to be sent to "words_topic" topic with the word
        KeyedMessage<String,String> data =
                new KeyedMessage<String,String>
                        ("words_topic",word);
        // Send the message
        producer.send(data);
    }
    System.out.println("Produced data");
    // close the producer
    producer.close();

}

// First paragraph from Franz Kafka's Metamorphosis
private static String MetaMORPHOSIS_opening_Para =
        "One morning,when Gregor Samsa woke from troubled dreams,"
                + "he found himself transformed in his bed into a horrible "
                + "vermin. He lay on his armour-like back,and if he lifted "
                + "his head a little he Could see his brown belly,slightly "
                + "domed and divided by arches into stiff sections.";

解决方法

我解决了

将Kafka代理的server.properties上的’advertised.host.name’设置为服务器的realIP(与生产者的’Metadata.broker.list’属性相同)

参考:https://issues.apache.org/jira/browse/KAFKA-1092

java Kafka生产者错误的更多相关文章

  1. HTML5 WebSocket实现点对点聊天的示例代码

    这篇文章主要介绍了HTML5 WebSocket实现点对点聊天的示例代码的相关资料,小编觉得挺不错的,现在分享给大家,也给大家做个参考。一起跟随小编过来看看吧

  2. ios – 在Swift的UIView中找到UILabel

    我正在尝试在我的UIViewControllers的超级视图中找到我的UILabels.这是我的代码:这是在Objective-C中推荐的方式,但是在Swift中我只得到UIViews和CALayer.我肯定在提供给这个方法的视图中有UILabel.我错过了什么?我的UIViewController中的调用:解决方法使用函数式编程概念可以更轻松地实现这一目标.

  3. ios – 在Swift中将输入字段字符串转换为Int

    所以我非常擅长制作APP广告Swift,我试图在文本字段中做一些非常简单的输入,取值,然后将它们用作Int进行某些计算.但是’vardistance’有些东西不正确它是导致错误的最后一行代码.它说致命错误:无法解开Optional.None解决方法在你的例子中,距离是一个Int?否则称为可选的Int..toInt()返回Int?因为从String到Int的转换可能失败.请参阅以下示例:

  4. 如何在iOS中检测文本(字符串)语言?

    例如,给定以下字符串:我想检测每个声明的字符串中使用的语言.让我们假设已实现函数的签名是:如果没有检测到语言,则返回可选字符串.因此,适当的结果将是:有一个简单的方法来实现它吗?

  5. xamarin – 崩溃在AccountStore.Create().保存(e.Account,“);

    在Xamarin.Forms示例TodoAwsAuth中https://developer.xamarin.com/guides/xamarin-forms/web-services/authentication/oauth/成功登录后,在aOnAuthenticationCompleted事件中,应用程序在尝试保存到Xamarin.Auth时崩溃错误说不能对钥匙串说期待着寻求帮助.解决方法看看你

  6. ios – 将视频分享到Facebook

    我正在编写一个简单的测试应用程序,用于将视频从iOS上传到Facebook.由于FacebookSDK的所有文档都在Objective-C中,因此我发现很难在线找到有关如何使用Swift执行此操作的示例/教程.到目前为止我有这个在我的UI上放置一个共享按钮,但它看起来已禁用,从我读到的这是因为没有内容设置,但我看不出这是怎么可能的.我的getVideoURL()函数返回一个NSURL,它肯定包含视

  7. xcode – 错误“线程1:断点2.1”

    我正在研究RESTAPI管理器.这是一个错误,我无法解决它.我得到的错误在下面突出显示.当我打电话给这个班级获取资源时:我评论的线打印:Thread1:breakpoint2.1我需要修复错误的建议.任何建议都非常感谢解决方法您可能在不注意的情况下意外设置了断点.单击并拖动代表断路器外部断点的蓝色刻度线以将其擦除.

  8. ios – 更改导航栏标题swift中的字符间距

    类型的值有人可以帮我这个或建议一种不同的方式来改变swift中导航栏标题中的字符间距吗?解决方法您无法直接设置属性字符串.你可以通过替换titleView来做一个技巧

  9. ios – 如何从变量访问属性或方法?

    是否可以使用变量作为Swift中方法或属性的名称来访问方法或属性?在PHP中,您可以使用$object->{$variable}.例如编辑:这是我正在使用的实际代码:解决方法你可以做到,但不能使用“纯粹的”Swift.Swift的重点是防止这种危险的动态属性访问.你必须使用Cocoa的Key-ValueCoding功能:非常方便,它完全穿过你要穿过的字符串到属性名称的桥,但要注意:这里是龙.

  10. ios – 如果我将自动释放的对象桥接到Core Foundation,我必须使用__bridge或__bridge_retained吗?

    ARC迁移工具遇到了这个问题:特别是,它不确定它是否应该执行__bridge或__bridge_retained.而我也是.-fileURLWithPath返回一个自动释放的对象,在这个地方我不是fileURL的所有者.但与此同时,该对象的保留计数至少为1.我敢打赌,这只能用__bridge来完成.解决方法您只想为此使用常规__bridge强制转换.仅当您想要管理强制转换CF对象的生命周期时,才会使用__bridge_retained.例如:所以__bridge_retained确实告诉编译器你有一个AR

随机推荐

  1. java – 合并多个相同的Kafka Streams主题

    )添加自定义存储,以使重复过滤容错.

  2. java – 我在哪里可以找到kafka的maven存储库?

    我想尝试kafka0.8.但是我在哪里可以找到kafkamaven存储库.我应该添加哪些额外的存储库URL?我找到了一些博客但它不起作用.我正在寻找合适的maven依赖.或者我应该从git中检出它并部署在我们的内部神器中?

  3. java – 为kafka主题配置ACL

    我有一个不安全的kafka实例,有2个代理,一切运行正常,直到我决定为主题配置ACL,在ACL配置后,我的消费者停止从Kafka轮询数据,并且在获取具有相关ID的元数据时我不断收到警告错误,我的代理属性看起来像下面:-我的客户端配置如下所示:–我用下面的命令来配置ACL在我启动消费者后完成所有上述配置后,它停止接收消息.有人可以指出我在误解的地方.提前致谢.解决方法我们成功使用ACL,但没有使用P

  4. java – kafka如何平衡分区加载?

    我在kafka遇到了负载均衡的问题.所以,我创建了一个包含10个分区的主题并创建了2个消费者.这10个分区被划分并分配给这些消费者,并且工作正常.有时第一个消费者工作,有时第二.但是,在某一时刻,我们可能面临一种情况,例如第二个消费者收到消息并且需要时间来处理此消息.那么,我的问题是kafka将如何决定将消息存储在哪个分区中?

  5. java – Kafka Connect实现错误

    我在这里运行教程:http://kafka.apache.org/documentation.html#introduction当我进入“步骤7:使用KafkaConnect导入/导出数据”并尝试启动两个连接器时,我收到以下错误:这是教程的一部分:Next,we’llstarttwoconnectorsrunninginstandalonemode,whichmeanstheyruninasing

  6. java – Storm-Kafka多个出口,如何共享负载?

    提前感谢你的时间.更新响应答案:现在在卡夫卡使用多分区(即5)以下是使用的代码:builder.setspout;在每个分区上用800MB数据进行淹水测试,花费22秒完成阅读.再次使用parallelism_hint=1的代码即builder.setspout;现在需要更多23秒!可以看到,这里的经纪人可以使用hosts.add添加,而且在新的KafkaConfig.StaticHosts代码片段中将partion号指定为4.如何提及卡夫卡喷嘴的并行提示?我是新来的风暴和java!!!!

  7. java Kafka生产者错误

    我做了kafkajava生产者.但控制台说错误.kafka服务器在aws上.和制作人在我的Mac上.然而kara服务器是可以访问的.当我从制作人发送消息时,kafka服务器显示“已接受的连接…”

  8. java – Kafka – 使用高级消费者的延迟队列实现

    如果是,我将消费两次相同的消息>问题等待很长时间没有提交偏移量,例如延迟时间是24小时,将从迭代器获取下一个,睡眠24小时,进程和提交(ZK会话超时?)>ZK会话如何在没有提交新的偏移量的情况下保持活动?

  9. java – 如何以健壮的方式处理kafka发布失败

    我正在使用Kafka,我们有一个用例来构建一个容错系统,甚至连一条消息都不会错过.所以这就是问题所在:如果由于任何原因(ZooKeeperdown,Kafkabroker等)向Kafka发布失败,我们如何能够有效地处理这些消息并在事情再次恢复后重播它们.正如我所说的那样,即使单个消息失败也无法承受.另一个用例是我们还需要在任何给定时间点知道有多少消息由于任何原因而无法发布到Kafka,例如计数器功

  10. java – Kafka快速入门:我需要哪些依赖项?

    接下来我该怎么办?我需要其他依赖项吗?解决方法问题是kafkabetawasbuiltinawaythatpomgeneratedwithajarisn’tvalidandmavencouldnotrecognizeitandparseproperly,从而获取传递依赖.我们已经设法通过在我们的pom定义中从该pom中获取所有依赖项来缓解此问题.我们正在等待kafka的下一个beta版本,其中将解决问题.完整依赖列表如下.请注意,您必须相应地更改scala版本依赖关系到kafka工件的后缀.至于Maybe

返回
顶部