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