Apache Avro Basics

Apache Avro Schema

./src/main/avro/com/cloudurable/phonebook/Employee.avsc

Example schema for an Employee record

{"namespace": "com.cloudurable.phonebook",

  "type": "record",
  "name": "Employee",

    "fields": [

        {"name": "firstName", "type": "string"},

        {"name": "lastName", "type": "string"},

        {"name": "age",  "type": "int"},

        {"name": "phoneNumber",  "type": "string"}
  
    ]
}

Avro schema generation tools

build.gradle - example using gradle-avro-plugin

plugins {
    id "com.commercehub.gradle.plugin.avro" version "0.9.0"
}

group 'cloudurable'
version '1.0-SNAPSHOT'
apply plugin: 'java'
sourceCompatibility = 1.8

dependencies {
    compile "org.apache.avro:avro:1.8.1"
    testCompile group: 'junit', name: 'junit', version: '4.11'
}

repositories {
    jcenter()
    mavenCentral()
}

avro {
    createSetters = false
    fieldVisibility = "PRIVATE"
}

./build/generated-main-avro-java/com/cloudurable/phonebook/Employee.java

Generated Avro code

package com.cloudurable.phonebook;

import org.apache.avro.specific.SpecificData;

@SuppressWarnings("all")
@org.apache.avro.specific.AvroGenerated
public class Employee extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
  private static final long serialVersionUID = -6112285611684054927L;
  public static final org.apache.avro.Schema SCHEMA$ = new    
                        org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"Employee\"...");
  public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
  private java.lang.String firstName;
  private java.lang.String lastName;
  private int age;
  private java.lang.String phoneNumber;
  ...

Using the new Employee class

Employee bob = Employee.newBuilder().setAge(35)
        .setFirstName("Bob")
        .setLastName("Jones")
        .setPhoneNumber("555-555-1212")
        .build();

assertEquals("Bob", bob.getFirstName());

Writing a list of employees to an Avro file

final List<Employee> employeeList = ...
final DatumWriter<Employee> datumWriter = new SpecificDatumWriter<>(Employee.class);
final DataFileWriter<Employee> dataFileWriter = new DataFileWriter<>(datumWriter);

try {
    dataFileWriter.create(employeeList.get(0).getSchema(),
            new File("employees.avro"));
    employeeList.forEach(employee -> {
        try {
            dataFileWriter.append(employee);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

    });
} finally {
    dataFileWriter.close();
}

Reading a list of employees from an avro file

final File file = new File("employees.avro");
final List<Employee> employeeList = new ArrayList<>();
final DatumReader<Employee> empReader = new SpecificDatumReader<>(Employee.class);
final DataFileReader<Employee> dataFileReader = new DataFileReader<>(file, empReader);

while (dataFileReader.hasNext()) {
    employeeList.add(dataFileReader.next(new Employee()));
}

Reading a list of employees from an avro file using forEach

final DataFileReader<Employee> dataFileReader = new DataFileReader<>(file, empReader);

dataFileReader.forEach(employeeList::add);

Using GenericRecord to create an Employee record

final String schemaLoc = "src/main/avro/com/cloudurable/phonebook/Employee.avsc";
final File schemaFile = new File(schemaLoc);
final Schema schema = new Schema.Parser().parse(schemaFile);

GenericRecord bob = new GenericData.Record(schema);
bob.put("firstName", "Bob");
bob.put("lastName", "Smith");
bob.put("age", 35);
assertEquals("Bob", bob.get("firstName"));

Writing GenericRecords to an Avro file

final List<GenericRecord> employeeList = new ArrayList<>();


final DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<>(schema);
final DataFileWriter<GenericRecord> dataFileWriter = new DataFileWriter<>(datumWriter);

try {
    dataFileWriter.create(employeeList.get(0).getSchema(),
            new File("employees2.avro"));
    employeeList.forEach(employee -> {
        try {
            dataFileWriter.append(employee);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    });
} finally {
    dataFileWriter.close();
}

Reading GenericRecords from an Avro file

final File file = new File("employees2.avro");
final List<GenericRecord> employeeList = new ArrayList<>();
final DatumReader<GenericRecord> empReader = new GenericDatumReader<>();
final DataFileReader<GenericRecord> dataFileReader = new DataFileReader<>(file, empReader);

while (dataFileReader.hasNext()) {
    employeeList.add(dataFileReader.next(null));
}

employeeList.forEach(System.out::println);

Using the wrong type

GenericRecord employee = new GenericData.Record(schema);
employee.put("firstName", "Bob" + index);
employee.put("lastName", "Smith"+ index);
//employee.put("age", index % 35 + 25);
employee.put("age", "OLD");

Stack trace from above

org.apache.avro.file.DataFileWriter$AppendWriteException: java.lang.ClassCastException:
java.lang.String cannot be cast to java.lang.Number

 at org.apache.avro.file.DataFileWriter.append(DataFileWriter.java:308)
 at com.cloudurable.phonebook.EmployeeTestNoGen.lambda$testWrite$1(EmployeeTestNoGen.java:71)
 at java.util.ArrayList.forEach(ArrayList.java:1249)
 at com.cloudurable.phonebook.EmployeeTestNoGen.testWrite(EmployeeTestNoGen.java:69)
 ...
Caused by: java.lang.ClassCastException: java.lang.String cannot be cast to java.lang.Number
 at org.apache.avro.generic.GenericDatumWriter.writeWithoutConversion(GenericDatumWriter.java:117)
 at org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:73)
 at org.apache.avro.generic.GenericDatumWriter.writeField(GenericDatumWriter.java:153)
 at org.apache.avro.generic.GenericDatumWriter.writeRecord(GenericDatumWriter.java:143)
 at org.apache.avro.generic.GenericDatumWriter.writeWithoutConversion(GenericDatumWriter.java:105)
 at org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:73)
 at org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:60)
 at org.apache.avro.file.DataFileWriter.append(DataFileWriter.java:302)

Stack trace from leaving out firstName

Caused by: java.lang.NullPointerException: null of string in field firstName of com.cloudurable.phonebook.Employee
 at org.apache.avro.generic.GenericDatumWriter.npe(GenericDatumWriter.java:132)
 at org.apache.avro.generic.GenericDatumWriter.writeWithoutConversion(GenericDatumWriter.java:126)
 at org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:73)
 at org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:60)

 {"namespace": "com.cloudurable.phonebook",
  "type": "record",
  "name": "Employee",
  "fields": [
    {"name": "firstName", "type": "string"},
    {"name": "nickName", "type": ["null", "string"], "default" : null},
    {"name": "lastName", "type": "string"},
    {"name": "age",  "type": "int"},
    {"name": "emails", "default":[], "type":{"type": "array", "items": "string"}},
    {"name": "phoneNumber",  "type":
      [ "null",
        { "type": "record",   "name": "PhoneNumber",
        "fields": [
          {"name": "areaCode", "type": "string"},
          {"name": "countryCode", "type": "string", "default" : ""},
          {"name": "prefix", "type": "string"},
          {"name": "number", "type": "string"}
        ]
        }
      ]
    },
    {"name":"status", "default" :"SALARY", "type": { "type": "enum", "name": "Status",
              "symbols" : ["RETIRED", "SALARY", "HOURLY", "PART_TIME"]}
    }
  ]
}

PhoneNumber record

package com.cloudurable.phonebook;

import org.apache.avro.specific.SpecificData;

@SuppressWarnings("all")
@org.apache.avro.specific.AvroGenerated
public class PhoneNumber extends org.apache.avro.specific.SpecificRecordBase ...{
  private static final long serialVersionUID = -3138777939618426199L;
  public static final org.apache.avro.Schema SCHEMA$ =
                   new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":...
  public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
   private java.lang.String areaCode;
   private java.lang.String countryCode;
   private java.lang.String prefix;
   private java.lang.String number;

Status enum

package com.cloudurable.phonebook;
@SuppressWarnings("all")
@org.apache.avro.specific.AvroGenerated
public enum Status {
  RETIRED, SALARY, HOURLY, PART_TIME  ;
  ...

Reakt, Java reactive programming lib, supports invokable promises so service methods could return a Promise.

Invokable Promise

        employeeService.lookupEmployee("123")
               .then((employee)-> {...}).catchError(...).invoke();

NOTE: QBit Microservices Lib version 1.4 (coming soon) will support generation of client stubs (local and remote) that return Reakt Promises. Having the client proxy method return the promise instead of taking a callback. With QBit you will be able to use Callback's on the service side and Promises on the client proxy side. Callbacks are natural for the service, and Promises are natural for the client.

But you do not have to use QBit to use Reakt's invokable promises.

Let's walk through an example of using a Reakt invokable promises.

Let's say we are developing a service discovery service with this interface.

ServiceDiscovery interface (Example service that does lookup)

    interface ServiceDiscovery {
        Promise<URI> lookupService(URI uri);

        default void shutdown() {}
        default void start() {}
    }

Note that ServiceDiscovery.lookupService is just an example likeEmployeeService.lookupEmployee was just an example.

Notice the lookupService returns a Promise (Promise<URI> lookupService(URI uri)). To call this, we can use an invokable promise. The service side will return an invokable promise, and the client of the service will use that Promise to register its then, thenExpect, thenMap, and/orcatchError handlers.

Here is the service side of the invokable promise example.

Service Side of invokable promise

    class ServiceDiscoveryImpl implements ServiceDiscovery {

        @Override
        public Promise<URI> lookupService(URI uri) {
            return invokablePromise(promise -> {
                if (uri == null) {
                    promise.reject("URI was null");
                } else {
                    ...
                    // DO SOME ASYNC OPERATION WHEN IT RETURNS CALL RESOLVE.
                    promise.resolve(successResult);
                }
            });
        }
    }

Notice to create the invokablePromise we use Promises.invokablePromise which takes aPromise Consumer (static <T> Promise<T> invokablePromise(Consumer<Promise<T>> promiseConsumer)).

The lookupService example returns a Promise and it does not execute its body until theinvoke (Promise.invoke) on the client side of the equation is called.

Let's look at the client side.

Client side of using lookupService.

        serviceDiscovery.lookupService(empURI)
                .then(this::handleSuccess)
                .catchError(this::handleError)
                .invoke();

The syntax this::handleSuccess is a method reference which can be used as Java 8 lambda expressions. We use method references to make the example shorter.

Here is a complete example with two versions of our example service.

Complete example from unit tests

package io.advantageous.reakt.promise;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.net.URI;
import java.util.Queue;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import static io.advantageous.reakt.promise.Promises.*;
import static org.junit.Assert.*;

public class InvokablePromise {


    final URI successResult = URI.create("http://localhost:8080/employeeService/");
    ServiceDiscovery serviceDiscovery;
    ServiceDiscovery asyncServiceDiscovery;
    URI empURI;
    CountDownLatch latch;
    AtomicReference<URI> returnValue;
    AtomicReference<Throwable> errorRef;

    @Before
    public void before() {
        latch = new CountDownLatch(1);
        returnValue = new AtomicReference<>();
        errorRef = new AtomicReference<>();
        serviceDiscovery = new ServiceDiscoveryImpl();
        asyncServiceDiscovery = new ServiceDiscoveryAsyncImpl();
        asyncServiceDiscovery.start();
        empURI = URI.create("marathon://default/employeeService?env=staging");
    }

    @After
    public void after() {
        asyncServiceDiscovery.shutdown();
    }

    public void await() {
        try {
            latch.await(10, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            throw new IllegalStateException(e);
        }
    }

    @Test
    public void testServiceWithReturnPromiseSuccess() {
        serviceDiscovery.lookupService(empURI).then(this::handleSuccess)
                .catchError(this::handleError).invoke();
        await();
        assertNotNull("We have a return", returnValue.get());
        assertNull("There were no errors", errorRef.get());
        assertEquals("The result is the expected result", successResult, returnValue.get());
    }


    @Test
    public void testServiceWithReturnPromiseFail() {


        serviceDiscovery.lookupService(null).then(this::handleSuccess)
                .catchError(this::handleError).invoke();

        await();
        assertNull("We do not have a return", returnValue.get());
        assertNotNull("There were  errors", errorRef.get());
    }


    @Test
    public void testAsyncServiceWithReturnPromiseSuccess() {
        asyncServiceDiscovery.lookupService(empURI).then(this::handleSuccess)
                .catchError(this::handleError).invoke();
        await();
        assertNotNull("We have a return from async", returnValue.get());
        assertNull("There were no errors form async", errorRef.get());
        assertEquals("The result is the expected result form async", successResult, returnValue.get());
    }


    @Test
    public void testAsyncServiceWithReturnPromiseFail() {


        asyncServiceDiscovery.lookupService(null).then(this::handleSuccess)
                .catchError(this::handleError).invoke();

        await();
        assertNull("We do not have a return from async", returnValue.get());
        assertNotNull("There were  errors from async", errorRef.get());
    }

    @Test (expected = IllegalStateException.class)
    public void testServiceWithReturnPromiseSuccessInvokeTwice() {
        final Promise<URI> promise = serviceDiscovery.lookupService(empURI).then(this::handleSuccess)
                .catchError(this::handleError);
        promise.invoke();
        promise.invoke();
    }

    @Test
    public void testIsInvokable() {
        final Promise<URI> promise = serviceDiscovery.lookupService(empURI).then(this::handleSuccess)
                .catchError(this::handleError);

        assertTrue("Is this an invokable promise", promise.isInvokable());
    }


    private void handleError(Throwable error) {
        errorRef.set(error);
        latch.countDown();
    }

    private void handleSuccess(URI uri) {
        returnValue.set(uri);
        latch.countDown();
    }


    interface ServiceDiscovery {
        Promise<URI> lookupService(URI uri);

        default void shutdown() {
        }

        default void start() {
        }
    }

    class ServiceDiscoveryImpl implements ServiceDiscovery {

        @Override
        public Promise<URI> lookupService(URI uri) {
            return invokablePromise(promise -> {

                if (uri == null) {
                    promise.reject("URI was null");
                } else {
                    promise.resolve(successResult);
                }
            });
        }
    }


    class ServiceDiscoveryAsyncImpl implements ServiceDiscovery {

        final ExecutorService executorService;

        final Queue<Runnable> runnables;

        final AtomicBoolean stop;

        public ServiceDiscoveryAsyncImpl() {
            executorService = Executors.newSingleThreadExecutor();
            runnables = new LinkedTransferQueue<>();
            stop = new AtomicBoolean();
        }

        @Override
        public Promise<URI> lookupService(URI uri) {
            return invokablePromise(promise -> {
                runnables.offer(() -> {
                    if (uri == null) {
                        promise.reject("URI was null");
                    } else {
                        promise.resolve(URI.create("http://localhost:8080/employeeService/"));
                    }
                });
            });
        }

        @Override
        public void shutdown() {
            stop.set(true);
            executorService.shutdown();
        }

        @Override
        public void start() {
            executorService.submit((Runnable) () -> {

                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                while (true) {
                    if (stop.get())break;
                    Runnable runnable = runnables.poll();
                    while (runnable != null) {
                        runnable.run();
                        runnable = runnables.poll();
                    }
                }

            });
        }
    }
}

<dependency>
    <groupId>io.advantageous</groupId>
    <artifactId>reakt</artifactId>
    <version>2.0.0.RELEASE</version>
</dependency>

compile 'io.advantageous:reakt:2.0.0.RELEASE'

  Promise<Employee> promise = promise()
                .then(e -> saveEmployee(e))
                .catchError(error -> 
                     logger.error("Unable to lookup employee", error));

  employeeService.lookupEmployee(33, promise);

  employeeService.lookupEmployee(33, 
        promise().then(e -> saveEmployee(e))
                 .catchError(error -> logger.error(
                                           "Unable to lookup ", error))
        );

        employeeService.lookupEmployee(33, result -> {
            result.then(e -> saveEmployee(e))
                  .catchError(error -> {
                    logger.error("Unable to lookup", error);
            });
        });

   public void lookupEmployee(long employeeId, Callback<Employee> callback){...}

        Promise<Employee> employeePromise = Promises.<Employee>blockingPromise();

        Promise<Sheep> sheepPromise = employeePromise
                .thenMap(employee1 -> new Sheep(employee1.getId()));

register(session.executeAsync("SELECT release_version FROM system.local"), 
  promise().thenExpect(expected -> 
     gui.setMessage("Cassandra version is " +
         expected.get().one().getString("release_version"))
  ).catchError(error -> 
     gui.setMessage("Error while reading Cassandra version: " 
     + error.getMessage())
  )
);

Czar Maker is a nice set of interfaces for Leader Election.

Czar Maker Consul is a Java lib for leadership election. Czar Maker Consul uses Consul to do leadership election.

<dependency>
    <groupId>io.advantageous.czarmaker</groupId>
    <artifactId>czar-maker-consul</artifactId>
    <version>0.1.0.RELEASE</version>
</dependency>

compile 'io.advantageous.czarmaker:czar-maker-consul:0.1.0.RELEASE'

import io.advantageous.consul.Consul;
import io.advantageous.czarmaker.Endpoint;
import io.advantageous.czarmaker.consul.*;
import io.advantageous.qbit.util.TestTimer;
import io.advantageous.reakt.promise.Promise;
import io.advantageous.reakt.promise.Promises;
import io.advantageous.reakt.reactor.Reactor;
import io.advantageous.reakt.reactor.TimeSource;

...

    private final long sessionTTL = 10;
    private final long newLeaderCheckInterval = 5;
    private ConsulLeadershipElector leadershipElector;
    private Reactor reactor;
    private TestTimer testTimer;
    private Consul consul;

...
        consul = Consul.consul();
        testTimer = new TestTimer();
        testTimer.setTime();
        reactor = Reactor.reactor(Duration.ofSeconds(30), new TestTimeSource(testTimer));
        final String serviceName = "foo";

        ConsulLeadershipProvider provider = new ConsulLeadershipProvider(serviceName, consul, TimeUnit.SECONDS, sessionTTL);

        leadershipElector = new ConsulLeadershipElector(provider, serviceName, reactor, TimeUnit.SECONDS,
                sessionTTL, newLeaderCheckInterval);


        /** Get the current leader. */
        Promise<Endpoint> promise = Promises.<Endpoint>blockingPromise();
        leadershipElector.getLeader(promise);

        assertTrue(promise.expect().isEmpty());


        /** Elect this endpoint as the current leader. */
        Promise<Boolean> selfElectPromise = Promises.<Boolean>blockingPromise();
        leadershipElector.selfElect(new Endpoint("foo.com", 9091), selfElectPromise);

        assertTrue("We are now the leader", selfElectPromise.get());


        /** Get the current leader again.  */
        Promise<Endpoint> getLeaderPromise = Promises.<Endpoint>blockingPromise();
        leadershipElector.getLeader(getLeaderPromise);

        /** See if it present. */
        assertTrue(getLeaderPromise.expect().isPresent());

        /** See if it has the host foo.com. */
        assertEquals("foo.com", getLeaderPromise.get().getHost());

        /** See if the port is 9091. */
        assertEquals(9091, getLeaderPromise.get().getPort());

        testTimer.seconds(100);

        leadershipElector.process();

        /** Elect a new leader. */
        leadershipElector.selfElect(new Endpoint("foo2.com", 9092), selfElectPromise);

QBit Release 1.0.0.RELEASE

Support was added for Consul Session to support leader election.

All of add-on libs into QBit Extensions and now the core is a lot smaller.

There was a flurry on new releases to improve the core RPC WebSocket lib.