作者热门文章
- android - RelativeLayout 背景可绘制重叠内容
- android - 如何链接 cpufeatures lib 以获取 native android 库?
- java - OnItemClickListener 不起作用,但 OnLongItemClickListener 在自定义 ListView 中起作用
- java - Android 文件转字符串
我的Kafka消费者代码如下,我只有一个消费者!
Properties consumerConfig = new Properties();
consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "sandbox.hortonworks.com:6667");
consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
consumerConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringDeserializer");
consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringDeserializer");
@SuppressWarnings("resource")
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(consumerConfig);
TestConsumerRebalanceListener rebalanceListener = new TestConsumerRebalanceListener();
consumer.subscribe(Collections.singletonList("test1"), rebalanceListener);
HDFSAppendTrial example = new HDFSAppendTrial();
String coreSite = "/usr/hdp/2.6.0.3-8/hadoop/etc/hadoop/core-site.xml";
String hdfsSite = "/usr/hdp/2.6.0.3-8/hadoop/etc/hadoop/hdfs-site.xml";
String hdfsFilePath = "/appendTo/Trial.csv";
while (true) {
ConsumerRecords<String, String> records = consumer.poll(999999999);
for (ConsumerRecord<String, String> record : records) {
FileSystem fileSystem = example.configureFileSystem(coreSite, hdfsSite);
String res = example.appendToFile(fileSystem, record.value(), hdfsFilePath);
System.out.printf("%s\n", record.value());
if (res.equalsIgnoreCase( "success")) {
System.out.println("Successfully appended to file");
}
else
System.out.println("couldn't append to file");
example.closeFileSystem(fileSystem);
}
consumer.commitSync();
}
在流式传输一些记录后出现以下错误
如何解决上述问题并解释问题的原因,如何克服!谁能帮我编写出满足我的更多要求的消费者代码。
提前感谢可以帮助我的人,也感谢至少尝试过的人
最佳答案
consumerConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
consumerConfig.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"True");
consumerConfig.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG,"99998");
consumerConfig.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG,"10000");
consumerConfig.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG,"99999");
我能够通过在属性对象 consumerConfig 中添加上述属性来解决此问题
关于java - 由于重新平衡,无法完成提交,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/45715126/
我是一名优秀的程序员,十分优秀!