DummyLog Producer Client not working..Trying to make DummyLog Example working

[Sandeep Mukhopadhyay]

Hi

I am trying to make Dummy Log Example working. Its giving some Avro Exception.

I am looking for working-example of camus (Avro to Avro ).  Please help me out

Code Snippet for DummyLogKafkaProducerClient

#####################################################333

package com.linkedin.camus.example.schemaregistry;

import java.util.Date;

import java.util.HashMap;

import java.util.Map;

import java.util.Properties;

import java.util.Random;

import kafka.javaapi.producer.Producer;

import kafka.producer.KeyedMessage;

import kafka.producer.ProducerConfig;

import com.linkedin.camus.etl.kafka.coders.KafkaAvroMessageEncoder;

import com.linkedin.camus.example.records.DummyLog;

public class DummyLogKafkaProducerClient {

    

    

    public static void main(String[] args) {

        Properties props = new Properties();

        props.put("metadata.broker.list", "localhost:6667");

        // props.put("serializer.class", "kafka.serializer.StringEncoder");

        // props.put("partitioner.class", "example.producer.SimplePartitioner");

        //props.put("request.required.acks", "1");

        ProducerConfig config = new ProducerConfig(props);

        Producer<String, byte[]> producer = new Producer<String, byte[]>(config);

        KafkaAvroMessageEncoder encoder = get_DUMMY_LOG_Encoder();

        for (int i = 0; i < 500; i++) {

            KeyedMessage<String, byte[]> data = new KeyedMessage<String, byte[]>("DUMMY_LOG", encoder.toBytes(getDummyLog()));

            producer.send(data);

        }

    }

    

    public static DummyLog getDummyLog() {

        Random random = new Random();

        DummyLog dummyLog = DummyLog.newBuilder().build();

        dummyLog.setId(random.nextLong());

        dummyLog.setLogTime(new Date().getTime());

        Map<CharSequence, CharSequence> machoStuff = new HashMap<CharSequence, CharSequence>();

        machoStuff.put("macho1", "abcd");

        machoStuff.put("macho2", "xyz");

        dummyLog.setMuchoStuff(machoStuff);

        return dummyLog;

    }

    

    public static KafkaAvroMessageEncoder get_DUMMY_LOG_Encoder() {

        KafkaAvroMessageEncoder encoder = new KafkaAvroMessageEncoder("DUMMY_LOG", null);

        Properties props = new Properties();

        props.put(KafkaAvroMessageEncoder.KAFKA_MESSAGE_CODER_SCHEMA_REGISTRY_CLASS, "com.linkedin.camus.example.schemaregistry.DummySchemaRegistry");

        encoder.init(props, "DUMMY_LOG");

        return encoder;

    }

}

I am also added Default no-arg constructor ot DummySchemaRegistry as it was giving instantiation Exception

public class DummySchemaRegistry extends MemorySchemaRegistry<Schema> {

    public DummySchemaRegistry(Configuration conf) {

        super();

        super.register("DUMMY_LOG", DummyLog.newBuilder().build().getSchema());

        super.register("DUMMY_LOG_2", DummyLog2.newBuilder().build()

                .getSchema());

    }

//no-arg constructor

    public DummySchemaRegistry() {

        super();

        super.register("DUMMY_LOG", DummyLog.newBuilder().build().getSchema());

        super.register("DUMMY_LOG_2", DummyLog2.newBuilder().build().getSchema());

    }

}

########################Exception I am getting

Exception in thread "main" com.linkedin.camus.coders.MessageEncoderException: org.apache.avro.AvroRuntimeException: org.apache.avro.AvroRuntimeException: Field id type:LONG pos:0 not set and has no default value

at com.linkedin.camus.etl.kafka.coders.KafkaAvroMessageEncoder.init(KafkaAvroMessageEncoder.java:55)

at com.linkedin.camus.example.schemaregistry.DummyLogKafkaProducerClient.get_DUMMY_LOG_Encoder(DummyLogKafkaProducerClient.java:57)

at com.linkedin.camus.example.schemaregistry.DummyLogKafkaProducerClient.main(DummyLogKafkaProducerClient.java:32)

Caused by: org.apache.avro.AvroRuntimeException: org.apache.avro.AvroRuntimeException: Field id type:LONG pos:0 not set and has no default value

at com.linkedin.camus.example.records.DummyLog$Builder.build(DummyLog.java:214)

at com.linkedin.camus.example.schemaregistry.DummySchemaRegistry.<init>(DummySchemaRegistry.java:16)

at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)

at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)

at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)

at java.lang.reflect.Constructor.newInstance(Constructor.java:408)

at java.lang.Class.newInstance(Class.java:438)

at com.linkedin.camus.etl.kafka.coders.KafkaAvroMessageEncoder.init(KafkaAvroMessageEncoder.java:52)

... 2 more

Caused by: org.apache.avro.AvroRuntimeException: Field id type:LONG pos:0 not set and has no default value

at org.apache.avro.data.RecordBuilderBase.defaultValue(RecordBuilderBase.java:151)

at com.linkedin.camus.example.records.DummyLog$Builder.build(DummyLog.java:209)

... 9 more

Regards,

Sandeep Mukhopadhyay

[Ambarish Hazarnis]

I suppose it expects the Avro schema to have default values. I had changed my dummyLog.avsc to following and recompile-

{

  "namespace": "com.linkedin.camus.example.records",

  "type": "record",

  "name": "DummyLog",

  "doc": "Logs for not so important stuff.",

  "fields": [

   {

"name": "id", 

"type": "int",

"default": 0 

},

   {

"name": "logTime", 

"type": "int",

"default": 0

}

]

}

Let me know if it works for you.

Thanks,

Ambarish