Please make sure to depend on version 3.5 of malhar-kafka in pom.xml. This
operator is not in malhar-library, it's a separate module.
On Fri, Oct 7, 2016 at 11:01 AM, Jaspal Singh <jaspal.singh1...@gmail.com>
wrote:
> Hi Siyuan,
>
> I am using the same Kafka producer as you mentioned. But I am not seeing
> the AbstractKafkaOutputOperator in malhar library while import.
>
>
> Thanks!!
>
> On Fri, Oct 7, 2016 at 12:52 PM, hsy...@gmail.com <hsy...@gmail.com>
> wrote:
>
>> Also which kafka output operator you are using?
>> Please use org.apache.apex.malhar.kafka.AbstractOutputOperator instead
>> of com.datatorrent.contrib.kafka.AbstractOutputOperator.
>> Only the org.apache.apex.malhar.kafka.AbstractOutputOperator works with
>> MapR stream, the latter only works with kafka 0.8.* or 0.9
>>
>> Regards,
>> Siyuan
>>
>> On Fri, Oct 7, 2016 at 10:38 AM, hsy...@gmail.com <hsy...@gmail.com>
>> wrote:
>>
>>> Hey Jaspal,
>>>
>>> Did you add any code to existing KafkaSinglePortExactlyOnceOutputOperator
>>> from malhar? If so please make sure the producer you use here is
>>> org.apache.kafka.clients.producer.KafkaProducer instead of
>>> kafka.javaapi.producer.Producer. That is old api and that is not
>>> supported by MapR stream.
>>>
>>>
>>> Regards,
>>> Siyuan
>>>
>>> On Fri, Oct 7, 2016 at 9:01 AM, Jaspal Singh <jaspal.singh1...@gmail.com
>>> > wrote:
>>>
>>>> Thomas,
>>>>
>>>> Below is the operator implementation we are trying to run. This
>>>> operator is getting an object of tenant class from updtream operator.
>>>>
>>>> public class KafkaSinglePortExactlyOnceOutputOperator extends
>>>> AbstractKafkaOutputOperator {
>>>>
>>>> private static final Logger LOG =
>>>> LoggerFactory.getLogger(KafkaSinglePortExactlyOnceOutputOperator.class);
>>>>
>>>> public transient final DefaultInputPort<Tenant> in = new
>>>> DefaultInputPort<Tenant>() {
>>>>
>>>> Gson gson = new Gson();
>>>>
>>>> @Override
>>>> public void process(Tenant tenant) {
>>>>
>>>> try {
>>>> Producer<String, String> producer = getKafkaProducer();
>>>> //ObjectMapper mapper = new ObjectMapper();
>>>> long now = System.currentTimeMillis();
>>>> //Configuration conf = HBaseConfiguration.create();
>>>> //TenantDao dao = new TenantDao(conf);
>>>> //ArrayList<Put> puts = new ArrayList<>();
>>>> if (tenant != null) {
>>>> //Tenant tenant = tenant.next();
>>>> if (StringUtils.isNotEmpty(tenant.getGl())) {
>>>> producer.send(new ProducerRecord<String,
>>>> String>("/datalake/corporate/ses_dlpoc/stream_fabbddev/jsingh73/sample-stream:certUpdate",
>>>> tenant.getVolumeName(), gson.toJson(tenant)));
>>>> //puts.add(dao.mkPut(tenant));
>>>> } else {
>>>> producer.send(new ProducerRecord<String,
>>>> String>("/datalake/corporate/ses_dlpoc/stream_fabbddev/jsingh73/sample-stream:error",
>>>> tenant.getVolumeName(), gson.toJson(tenant)));
>>>>
>>>> }
>>>> producer.flush();
>>>> }
>>>> }
>>>>
>>>>
>>>> After building the application, it throws error during launch:
>>>>
>>>> An error occurred trying to launch the application. Server message:
>>>> java.lang.NoClassDefFoundError: Lkafka/javaapi/producer/Producer; at
>>>> java.lang.Class.getDeclaredFields0(Native Method) at
>>>> java.lang.Class.privateGetDeclaredFields(Class.java:2583) at
>>>> java.lang.Class.getDeclaredFields(Class.java:1916) at
>>>>
>>>>
>>>> Thanks
>>>> Jaspal
>>>>
>>>> On Fri, Oct 7, 2016 at 10:42 AM, Jaspal Singh <
>>>> jaspal.singh1...@gmail.com> wrote:
>>>>
>>>>> Thomas,
>>>>>
>>>>> I was trying to refer to the input from previous operator.
>>>>>
>>>>> Another thing when we extend the AbstractKafkaOutputOperator, do we
>>>>> need to specify <String, T> ? Since we are getting an object of class type
>>>>> from previous operator.
>>>>>
>>>>>
>>>>> Thanks
>>>>> Jaspal
>>>>>
>>>>> On Fri, Oct 7, 2016 at 10:12 AM, Thomas Weise <t...@apache.org> wrote:
>>>>>
>>>>>> Are you referring to the upstream operator in the DAG or the state of
>>>>>> the previous application after relaunch? Since the data is stored in MapR
>>>>>> streams, an operator that is a producer can also act as a consumer.
>>>>>> Please
>>>>>> clarify your question.
>>>>>>
>>>>>>
>>>>>> On Fri, Oct 7, 2016 at 7:59 AM, Jaspal Singh <
>>>>>> jaspal.singh1...@gmail.com> wrote:
>>>>>>
>>>>>>> Hi Thomas,
>>>>>>>
>>>>>>> I have a question, so when we are using
>>>>>>> *KafkaSinglePortExactlyOnceOutputOperator* to write results into
>>>>>>> maprstream topic will it be able to read messgaes from the previous
>>>>>>> operator ?
>>>>>>>
>>>>>>>
>>>>>>> Thanks
>>>>>>> Jaspal
>>>>>>>
>>>>>>> On Thu, Oct 6, 2016 at 6:28 PM, Thomas Weise <t...@apache.org> wrote:
>>>>>>>
>>>>>>>> For recovery you need to set the window data manager like so:
>>>>>>>>
>>>>>>>> https://github.com/DataTorrent/examples/blob/master/tutorial
>>>>>>>> s/exactly-once/src/main/java/com/example/myapexapp/Applicati
>>>>>>>> on.java#L33
>>>>>>>>
>>>>>>>> That will also apply to stateful restart of the entire application
>>>>>>>> (relaunch from previous instance's checkpointed state).
>>>>>>>>
>>>>>>>> For cold restart, you would need to consider the property you
>>>>>>>> mention and decide what is applicable to your use case.
>>>>>>>>
>>>>>>>> Thomas
>>>>>>>>
>>>>>>>>
>>>>>>>> On Thu, Oct 6, 2016 at 4:16 PM, Jaspal Singh <
>>>>>>>> jaspal.singh1...@gmail.com> wrote:
>>>>>>>>
>>>>>>>>> Ok now I get it. Thanks for the nice explaination !!
>>>>>>>>>
>>>>>>>>> One more thing, so you mentioned about checkpointing the offset
>>>>>>>>> ranges to replay in same order from kafka.
>>>>>>>>>
>>>>>>>>> Is there any property we need to configure to do that? like
>>>>>>>>> initialOffset set to APPLICATION_OR_LATEST.
>>>>>>>>>
>>>>>>>>>
>>>>>>>>> Thanks
>>>>>>>>> Jaspal
>>>>>>>>>
>>>>>>>>>
>>>>>>>>> On Thursday, October 6, 2016, Thomas Weise <thomas.we...@gmail.com>
>>>>>>>>> wrote:
>>>>>>>>>
>>>>>>>>>> What you want is the effect of exactly-once output (that's why we
>>>>>>>>>> call it also end-to-end exactly-once). There is no such thing as
>>>>>>>>>> exactly-once processing in a distributed system. In this case it
>>>>>>>>>> would be
>>>>>>>>>> rather "produce exactly-once. Upstream operators, on failure, will
>>>>>>>>>> recover
>>>>>>>>>> to checkpointed state and re-process the stream from there. This is
>>>>>>>>>> at-least-once, the default behavior. Because in the input operator
>>>>>>>>>> you have
>>>>>>>>>> configured to replay in the same order from Kafka (this is done by
>>>>>>>>>> checkpointing the offset ranges), the computation in the DAG is
>>>>>>>>>> idempotent
>>>>>>>>>> and the output operator can discard the results that were already
>>>>>>>>>> published
>>>>>>>>>> instead of producing duplicates.
>>>>>>>>>>
>>>>>>>>>> On Thu, Oct 6, 2016 at 3:57 PM, Jaspal Singh <
>>>>>>>>>> jaspal.singh1...@gmail.com> wrote:
>>>>>>>>>>
>>>>>>>>>>> I think this is something called a customized operator
>>>>>>>>>>> implementation that is taking care of exactly once processing at
>>>>>>>>>>> output.
>>>>>>>>>>>
>>>>>>>>>>> What if any previous operators fail ? How we can make sure they
>>>>>>>>>>> also recover using EXACTLY_ONCE processing mode ?
>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>>> Thanks
>>>>>>>>>>> Jaspal
>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>>> On Thursday, October 6, 2016, Thomas Weise <
>>>>>>>>>>> thomas.we...@gmail.com> wrote:
>>>>>>>>>>>
>>>>>>>>>>>> In that case please have a look at:
>>>>>>>>>>>>
>>>>>>>>>>>> https://github.com/apache/apex-malhar/blob/master/kafka/src/
>>>>>>>>>>>> main/java/org/apache/apex/malhar/kafka/KafkaSinglePortExactl
>>>>>>>>>>>> yOnceOutputOperator.java
>>>>>>>>>>>>
>>>>>>>>>>>> The operator will ensure that messages are not duplicated,
>>>>>>>>>>>> under the stated assumptions.
>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>> On Thu, Oct 6, 2016 at 3:37 PM, Jaspal Singh <
>>>>>>>>>>>> jaspal.singh1...@gmail.com> wrote:
>>>>>>>>>>>>
>>>>>>>>>>>>> Hi Thomas,
>>>>>>>>>>>>>
>>>>>>>>>>>>> In our case we are writing the results back to
>>>>>>>>>>>>> maprstreams topic based on some validations.
>>>>>>>>>>>>>
>>>>>>>>>>>>>
>>>>>>>>>>>>> Thanks
>>>>>>>>>>>>> Jaspal
>>>>>>>>>>>>>
>>>>>>>>>>>>>
>>>>>>>>>>>>> On Thursday, October 6, 2016, Thomas Weise <t...@apache.org>
>>>>>>>>>>>>> wrote:
>>>>>>>>>>>>>
>>>>>>>>>>>>>> Hi,
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> which operators in your application are writing to external
>>>>>>>>>>>>>> systems?
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> When you look at the example from the blog (
>>>>>>>>>>>>>> https://github.com/DataTorrent/examples/tree/master/tutoria
>>>>>>>>>>>>>> ls/exactly-once), there is Kafka input, which is configured
>>>>>>>>>>>>>> to be idempotent. The results are written to JDBC. That operator
>>>>>>>>>>>>>> by itself
>>>>>>>>>>>>>> supports exactly-once through transactions (in conjunction with
>>>>>>>>>>>>>> idempotent
>>>>>>>>>>>>>> input), hence there is no need to configure the processing mode
>>>>>>>>>>>>>> at all.
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> Thomas
>>>>>>>>>>>>>>
>>>>>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>
>>>>>>>
>>>>>>
>>>>>
>>>>
>>>
>>
>
