title: Kafka AVRO Consumer: MySQL Decimal to Java Decimal
body: <p>I'm trying to consume records from a MySQL table which contains 3 columns <code>(Axis, Price, lastname)</code> with their datatypes <code>(int, decimal(14,4), varchar(50))</code> respectively. </p>
  
  <p>I inserted one record which has the following data <code>(1, 5.0000, John)</code>.</p>
  
  <p>The following Java code (which consumes the AVRO records from a topic created by a MySQL Connector in Confluent platform) reads the decimal column: Price, as java.nio.HeapByteBuffer type so i can't reach the value of the column when i receive it.</p>
  
  <p>Is there a way to extract or convert the received data to a Java decimal or double data type?</p>
  
  <p>Here is the MySQL Connector properties file:-</p>
  
  <pre><code>{
    "name": "mysql-source",
    "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
     "key.converter": "io.confluent.connect.avro.AvroConverter",
     "key.converter.schema.registry.url": "http://localhost:8081",
     "value.converter": "io.confluent.connect.avro.AvroConverter",
     "value.converter.schema.registry.url": "http://localhost:8081",
     "incrementing.column.name": "Axis",
     "tasks.max": "1",
     "table.whitelist": "ticket",
     "mode": "incrementing",
     "topic.prefix": "mysql-",
     "name": "mysql-source",
     "validate.non.null": "false",
     "connection.url": "jdbc:mysql://localhost:3306/ticket? 
     user=user&amp;password=password"
     }
  }
  </code></pre>
  
  <p>Here is the code:- </p>
  
  <pre><code>    public static void main(String[] args) throws InterruptedException, 
       IOException {
  
          Properties props = new Properties();
  
          props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
  props.put(ConsumerConfig.GROUP_ID_CONFIG, "group1");
  
  
          props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
          "org.apache.kafka.common.serialization.StringDeserializer");
          props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
          "io.confluent.kafka.serializers.KafkaAvroDeserializer");
          props.put("schema.registry.url", "http://localhost:8081");
  
          props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
  
          String topic = "sql-ticket";
  final Consumer&lt;String, GenericRecord&gt; consumer = new KafkaConsumer&lt;String, GenericRecord&gt;(props);
  consumer.subscribe(Arrays.asList(topic));
  
  try {
    while (true) {
      ConsumerRecords&lt;String, GenericRecord&gt; records = consumer.poll(100);
      for (ConsumerRecord&lt;String, GenericRecord&gt; record : records) {
        System.out.printf("value = %s \n", record.value().get("Price"));
      }
    }
  } finally {
    consumer.close();
  }
  
  }
  </code></pre>
  
  <p><a href="https://i.stack.imgur.com/78BLs.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/78BLs.png" alt="enter image description here"></a></p>