[
https://issues.apache.org/jira/browse/HADOOP-14623?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16076628#comment-16076628
]
Hongyuan Li edited comment on HADOOP-14623 at 7/8/17 11:55 AM:
---------------------------------------------------------------
futuremore, flush method is to confirm that data has been written.
*Update/Crorrection*
sorry, it is the {{putMetrics}} method.
in {{KafkaSink}}#{{putMetrics}} , code lists below， which makes me have a
different opinion:
{code}
……
Future<RecordMetadata> future = producer.send(data);
jsonLines.setLength(0);
try {
future.get(); // which means synchronously
} catch (InterruptedException e) {
throw new MetricsException("Error sending data", e);
} catch (ExecutionException e) {
throw new MetricsException("Error sending data", e);
}
……
{code}
was (Author: hongyuan li):
futuremore, flush method is to confirm that data has been written.
*Update/Crorrection*
sorry, it is the {{putMetrics}} method.
in {{KafkaSink}}#{{putMetrics}} , code lists below， which makes me have a
different opinion:
{code}
……
Future<RecordMetadata> future = producer.send(data);
jsonLines.setLength(0);
try {
future.get();
} catch (InterruptedException e) {
throw new MetricsException("Error sending data", e);
} catch (ExecutionException e) {
throw new MetricsException("Error sending data", e);
}
……
{code}
> fixed some bugs in KafkaSink
> -----------------------------
>
> Key: HADOOP-14623
> URL: https://issues.apache.org/jira/browse/HADOOP-14623
> Project: Hadoop Common
> Issue Type: Bug
> Components: common, tools
> Affects Versions: 3.0.0-alpha3
> Reporter: Hongyuan Li
> Assignee: Hongyuan Li
> Attachments: HADOOP-14623-001.patch, HADOOP-14623-002.patch
>
>
> {{KafkaSink}}#{{init}} should set ack to *1* to make sure the message has
> been written to the broker at least.
> current code list below:
> {code}
>
> props.put("request.required.acks", "0");
> {code}
> *Update*
> find another bug about this class, {{key.serializer}} used
> {{org.apache.kafka.common.serialization.ByteArraySerializer}}, however, the
> key properties of Producer is Integer, codes list below:
> {code}
> props.put("key.serializer",
> "org.apache.kafka.common.serialization.ByteArraySerializer");
> ……………
> producer = new KafkaProducer<Integer, byte[]>(props);
> {code}
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
