Fury fury = Fury.builder().withLanguage(Fury.Language.JAVA)
         .withReferenceTracking(true).build();
SomeClass obj = new SomeClass();
byte[] bytes = fury.serialize(obj);
Object newobj = fury.deserialize(bytes);

SomeClass object = new SomeClass();
{
  Fury fury = Fury.builder().withLanguage(Fury.Language.JAVA)
      .withReferenceTracking(true).build();
  byte[] bytes = fury.serialize(object);
  Object o = fury.deserialize(bytes);
}
{
  ThreadSafeFury fury = Fury.builder().withLanguage(Fury.Language.JAVA)
      .withReferenceTracking(true).buildThreadSafeFury();
  byte[] bytes = fury.serialize(object);
  Object o = fury.deserialize(bytes);
}
{
  ThreadSafeFury fury = new ThreadSafeFury(() -> {
    Fury fury = Fury.builder().withLanguage(Fury.Language.JAVA)
        .withReferenceTracking(true).build();
    fury.register(xxx.class);
    return fury;
  });
  byte[] bytes = fury.serialize(object);
  Object o = fury.deserialize(bytes);
}

Foo foo = Foo.create();
Encoder<Foo> encoder = Encoders.bean(Foo.class);
int initSize = 512;
MemoryBuffer buffer = MemoryUtils.buffer(initSize);
int numRows = 128;

VectorSchemaRoot root = ArrowUtils.
  createVectorSchemaRoot(encoder.schema());
ArrowWriter arrowWriter = new ArrowWriter(root);
try (ArrowStreamWriter writer = new ArrowStreamWriter(
    root, null, new FuryOutputStream(buffer))) {
  writer.start();
  for (int i = 0; i < numRows; i++) {
    arrowWriter.write(encoder.toRow(foo));
  }
  arrowWriter.finish();
  writer.writeBatch();
  writer.end();
}

Foo foo = Foo.create();
Encoder<Foo> encoder = Encoders.bean(Foo.class);
int initSize = 512;
MemoryBuffer buffer = MemoryUtils.buffer(initSize);
int numRows = 128;

buffer.writerIndex(0);
ArrowWriter arrowWriter = ArrowUtils.createArrowWriter(encoder.schema());
for (int i = 0; i < numRows; i++) {
  BinaryRow row = encoder.toRow(foo);
  arrowWriter.write(row);
}
ArrowRecordBatch recordBatch = arrowWriter.finishAsRecordBatch();
DataTypes.serializeSchema(encoder.schema(), buffer);
ArrowUtils.serializeRecordBatch(recordBatch, buffer);
arrowWriter.reset();
ArrowStreamWriter.writeEndOfStream(new WriteChannel(Channels.newChannel(
    new FuryOutputStream(buffer))), new IpcOption());

public class Foo {
  public int f1;
  public String f2;
  public List<String> f3;
  public Map<String, Integer> f4;
  public Bar f45;
}

public class Bar {
  public int f1;
  public String f2;
}

Encoder<Foo> encoder = Encoders.encoder(Foo);

public byte[] write(Foo foo) {
  return encoder.toRow(foo).toBytes();
}

public Foo read(byte[] bytes) {
  BinaryRow newRow = new BinaryRow(encoder.schema());
  newRow.pointTo(bytes, 0, bytes.size());
  return encoder.fromRow(newRow);
}

public class Foo {
  public int f1;
  public Map<String, Bar> f2;
  public Long f3;
}

public class Bar {
  public String f1;
  public List<String> f2;
}

Encoder<Foo> encoder = Encoders
  .encoder(Foo.class);
byte[] bytes = encoder.toBytes(foo))

public static byte[] serialize(Foo foo) {
  Map<String, Generated.Bar> pbF2 =
      foo.f2.entrySet().stream()
      .collect(Collectors.toMap(
          Map.Entry::getKey, e -> {
        Bar bar = e.getValue();
        return Generated.Bar.newBuilder()
            .setF1(bar.f1).addAllF2(bar.f2)
            .build();
      }));
  return Generated.Foo.newBuilder()
      .setF1(foo.f1)
      .putAllF2(pbF2)
      .setF3(foo.f3)
      .build()
      .toByteArray();
}

public Foo deserialize(byte[] bytes) throws Exception {
  Generated.Foo pbFoo = Generated.Foo.parseFrom(bytes);
  Foo foo = new Foo();

  foo.f1 = pbFoo.getF1();
  foo.f2 = pbFoo.getF2Map().entrySet().stream().collect(
      Collectors.toMap(Map.Entry::getKey, e -> {
        Generated.Bar value = e.getValue();
        Bar bar = new Bar();
        bar.f1 = value.getF1();
        bar.f2 = new ArrayList<>(value.getF2List());
        return bar;
      }));
  foo.f3 = pbFoo.getF3();
  return foo;
}

// Normal task
RayObject<String> hello = Ray.call(MyClass::echo, "hello");
RayObject<String> res = Ray.call(
        new PyRemoteFunction<>("module", "func", String.class));

// Actor
ActorCreationOptions options = new ActorCreationOptions.Builder()
    .setMaxRestarts(-1).createActorCreationOptions();
RayActor<MyActor> myActor = Ray.createActor(MyActor::new, 10, options);
Integer result = Ray.call(MyActor::increaseAndGet, actor, 1).get();
RayPyActor actor = Ray.createActor(new PyActorClass("module", "func"));
RayObject<byte[]> res = Ray.call(new PyActorMethod<>("fetch", byte[].class), actor);

// Normal task
ObjectRef<String> obj = Ray.task(MyClass::echo, "hello").remote();
ObjectRef<String> res = Ray.task(
          PyFunction.of("module", "func", String.class)).remote();

// Actor
ActorHandle<MyActor> myActor = Ray.actor(MyActor::new, 10)
    .setMaxRestarts(-1).remote();
Integer obj = myActor.task(MyActor::increaseAndGet, 1).remote().get();
PyActorHandle pyActor = Ray.actor(PyActorClass.of("module", "Actor")).remote();
ObjectRef<byte[]> res = actor.task(PyActorMethod.of("fetch", byte[].class),

public Foo foo(Bar bar, Map<String, List<Integer>> map){
  return new Foo(bar, map);
}

Ray.task(PyFunction.of("ray_demo", "foo", Foo.class),
  bar, map).remote();
pyActorHandle.task(...).remote(bar, map);

public class Vertex {
  public int index;
  public Map<String, Double> props;
  public Vertex to
}
fury.registerClass(Vertex.class, "example.Vertex");
List<Vertex> vertices = Arrays.asList(create(), create());
vertices.get(0).to = vertices.get(1);
vertices.get(1).to = vertices.get(0);
byte[] bytes = fury.serialize(vertices));

public class Vertex {
  public int index;
  public Map<String, Double> props;
  public Vertex to
}
fury.registerClass(Vertex.class, "example.Vertex");
List<Vertex> vertices = fury.deserialize(binary));