Continuing on the Big Data theme, Google Cloud Dataflow is the next component I want to take a look in Google Cloud Platform.

What is Dataflow?

• Dataflow is mainly for batch or stream data processing.
• Good for high volume computation and embarrassingly parallel workloads.
• Consists of 2 major components:
  1. Dataflow SDKs: A programming model and SDKs for large-scale cloud data processing.
  2. Dataflow Service: Ties together and fully manages several different Google Cloud Platform technologies to execute data processing jobs in the cloud.
• Dataflow SDK is being open sourced as Apache Beam.

Dataflow Programming Model

Dataflow Programming Model consists of 4 concepts:

1. Pipelines: Set of operations that can read a source of input data, transform it and write out the output. Contains data (PCollections) and processing on the data (Transforms)
2. PCollections: Inputs and outputs for each step in the pipeline. Immutable after creation. 2 flavors:
   • Bounded: Fixed-size data set for text, BigQuery, Datastore or custom data.
   • Unbounded: Continuously updating data set, or streaming data such as Pub/Sub or custom data.
3. Transforms: A data processing operation, or a step, in the pipeline. Takes PCollection as input and produces PCollection as output. 2 flavors:
   • Core:  You provide the processing logic as a function object. 4 Core transform types: ParDo, GroupByKey, Combine, Flatten.
   • Composite: Built from multiple sub-transforms.
4. I/O Sources and Sinks: Source APIs to read data into the pipeline, and sink APIs to write output data from your pipeline. APIs for common formats such as:
   • Text files
   • BigQuery tables
   • Avro files
   • Pub/Sub
   • BigTable (beta)

Dataflow SDKs

Two supported languages:

1. Java: Dataflow SDK for Java is fully available
2. Python: Dataflow SDK for Python is in development.

Dataflow Service

• Dataflow Service is a managed service in Google Cloud Platform to deploy and execute Dataflow pipelines (as Dataflow jobs).
• Simplifies distributed parallel processing by:
  1. Automatic partitioning and distribution of Compute Engine instances.
  2. Optimization of the pipeline.
  3. Automatic scaling of resources as needed.
• Automatically spins and tears down necessary resources (Compute Engine, Cloud Storage) to run the Dataflow job.
• Provides tools like Dataflow Monitoring Interface and the Dataflow Command-line Interface.

Resources

• Dataflow Main Docs
• Dataflow Programming Model
• Dataflow Java SDK
• Dataflow Java SDK JavaDocs
• Dataflow Service Docs

Intro

The next component I want to explore in the Compute layer of Google Cloud Platform (GCP) is App Engine. As usual, App Engine Docs is the best place to learn more but I want to provide shorter notes here. Being a Java guy myself, I also want to emphasize aspects of App Engine relevant to Java.

What is App Engine?

• Platform as a Service (PaaS) offering of Google Cloud.
• Provides managed runtimes for specific versions of Java, Go, PHP, Python.
• Mainly for web and mobile apps.
• App Engine provides:
  • Automatic scaling and load balancing.
  • Access to other App Engine services such as Datastore, Blobstore, Memcache, Endpoints, Task Queues, Scheduled Tasks, Search API, Logs API and more.
• Each language comes with an App Engine SDK for local development, testing and deployment.
• There are 2 flavors of App Engine:
  1. Sandbox: Provides a sandboxed (and limited) environment to run your web app.
  2. Managed VM (beta): Built on top of Compute Engine, provides more freedom on environment.

Sandbox

In the default flavor of App Engine, your Java app needs to conform these sandbox constraints in order to run on App Engine:

• App needs to run in Java 7 and a limited subset of Java standard library.
• App cannot write to file system (use Datastore instead). 
• No arbitrary network connections.
• No slow responses: A web request must be handled within reasonable amount of time (the exact time depends on scaling mode).
• Limited threads and background threads (depends on scaling mode).

App Engine (Sandbox) and Java

• Runs your Java web application using Java 7 in a sandboxed environment.
• Uses the Java Servlet 2.5 standard.
• App Engine Java SDK for development. It also has a plugin for supporting development with Apache Maven.
• Development server for local development and testing.
• IDE support: Google Plugin for Eclipse to manage App Engine directly from Eclipse. Additionally, Google App Engine Integration for IntelliJ and NetBeans plugin for Gaelyk Framework.
• Java apps are deployed as App Engine Modules where each module can act like a microservice.
• In terms of configuration:
  • Top level app is packaged as EAR (similar to JEE).
  • Individual modules are packaged as WAR within EAR (again similar to JEE).
• Each Module can independently choose its own scaling type (how each instance is created and scaled)
  • 3 scaling types: Manual, Basic and Automatic.
• Each scaling type has its own instance classes. The instance class determines compute resources and pricing:
  • Manual: B1, B2, B4, B4_1G, or B8.
  • Basic: B1, B2, B4, B4_1G, or B8.
  • Automatic: F1, F2, F4, or F4_1G.

Managed VM

• Managed VM is in beta.
• Based on Google Compute Engine, allows one to customize runtime and OS (beyond what Sandbox provides) using Docker while providing the benefits of App Engine (automatic scaling, load balancing etc.)
• Native support for Java 8 / Servlet 3.1, Java 7 / Jetty 9, Python 2.7 and Python 3.4, Node.js, and Go. Further customization with Docker.
• Google Cloud SDK is needed to administer Managed VMs.

Resources

• Main App Engine Docs
• App Engine Features List
• App Engine (Sandbox) Docs for Java
• App Engine (Managed VM) Docs for Java 
• Google App Engine API for Java
• JRE Class White List
• App Engine Modules in Java
• App Engine Java Runtime

In this post, I want to take a look at the Java client libraries. There are 2 client libraries for Java:

1. Apache JClouds: This is an open source library that allows one to use Java across multiple Cloud providers, including Compute Engine on Google Cloud. This is a good option if you want to work with multiple cloud vendors from the same API.
2. Google API Client Library for Java: Provided by Google, this is a Java library for accessing all Google APIs (not just Google Cloud), including Compute Engine on Google Cloud.

A good starting point to explore Google API Client Library for Java is Compute Engine Cmdline Sample. This sample basically contains a single class ComputeEngineSample that lists all compute engine instances you have in your project from Java. Nothing earth-shattering but it shows how to authenticate with Google API and interact with Compute Engine.

There’s an instructions.html file that tells you how to setup and use the sample. One thing you’ll realize is that you need to grant API access to your Java app before you can run the sample but instructions.html is a little outdated on that. This is what you need to do to grant API access:

1. Either create or go to your project from Developer Console.
2. Open side-menu on top left and go to “API Manager”.
3.  Under “API Manager”, click on “Credentials” -> “Create Credentials” -> “OAuth Client ID”.
4. Now, you need to select the application type. In our case, select “Other” and give it a name.
5. At this point, you’ll have a pop-up with client ID and secret. Copy these down as you’ll need them later.

Once that’s done, you can import the project into Eclipse (or your favorite IDE), fill client_secrets.json in resources folder with client ID and secret from step 5, and fill the following info in ComputeEngineSample:

• APPLICATION_NAME: Optional but give it a name, so you don’t get warnings.

• projectId: Your project id from Developers Console.

• zoneName: Make sure to enter the right zone for your project. In my case, I chose europe-west1-c.

That’s it. Now, you can run ComputeEngineSample from Eclipse (or your favorite IDE) and you can get info about your Compute Engine instances right from Java:

================== Listing Compute Engine Instances ==================
{
"canIpForward" : false,
"creationTimestamp" : "2016-02-16T05:58:00.714-08:00",
"description" : "",
"disks" : [ {
"autoDelete" : true,
"boot" : true,
"deviceName" : "instance-1",
"index" : 0,
"kind" : "compute#attachedDisk",
"mode" : "READ_WRITE",
    ....

So, you might be tempted to mark a thread as daemon and blindly assume that it will never block the shutdown of your application. While this is true in most cases, as always, there are exceptions. Consider the following piece of code:

    public static void main(String[] args) {
        Thread t = new Thread(new Runnable() {
            @Override
            public void run() {
                int i = 0;
                while (true) {
                    System.out.println(i++);
                    try {
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });
        t.setDaemon(true);
        t.start();

        try {
            t.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("end");
    }

A daemon thread is created which simply prints a counter every second. Then, there’s a call to Thread#join to wait for the daemon thread to finish and then the main thread prints “end”. Well, “end” will never be printed and the application will not exit because the daemon thread will continue printing counter and Thread#join will block forever. This might be surprising at first glance. Isn’t this a daemon thread? Isn’t it supposed to not block JVM? Yes, but calling Thread#join on a daemon thread effectively turns the daemon thread into a normal thread.

The lesson here is that marking a thread as daemon is only part of the story. How a daemon thread is used throughout your application is also as important and you need to keep that in mind. Another lesson is that don’t use things that can block forever such as Thread#join. I know you designed your application so perfect that you think it will never block but if something can block in theory, it will block in production sooner or later. That can be quite annoying to customers and support people, so those people before blindly implementing blocking code paths. Instead use non-blocking alternatives like Thread#join(long) and provide a timeout.

In Java, both Thread.sleep and Object.wait make the current thread wait for a specified amount of time. This is useful when the current thread needs to wait for some other thread before it can proceed. I sometimes see developers use these interchangeably in close proximity in code but this can be problematic as the two methods are quite different.

Consider the example from my previous post. In that example, thread1 acquired a lock, slept for a minute while thread2 was blocked waiting for the same lock. It used TimeUnit.MINUTES.sleep(1) which uses Thread.sleep to make thread1 wait. Let’s change it to use Object.wait and see what happens:

private static void test() {
    final Object lock = new Object();

    Thread thread1 = new Thread(new Runnable() {
        @Override public void run() {
            synchronized (lock) {
                System.out.println("Thread1 acquired lock");
                try {
                    lock.wait(1 * 60 * 1000);
                    //TimeUnit.MINUTES.sleep(1);
                } catch (InterruptedException ignore) {}
            }
        }

    });
    thread1.start();

    Thread thread2 = new Thread(new Runnable() {
        @Override public void run() {
            synchronized (lock) {
                System.out.println("Thread2 acquired lock");
            }
        }
    });
    thread2.start();
}

Once you run this sample, you’ll see the following right away:

Thread1 acquired lock
Thread2 acquired lock

So what happened? When thread1 went to sleep, it released the lock and thread2 acquired it right away. Unlike Thread.sleep (where locks are not released) in Object.wait, locks are released as the thread goes to sleep. In some situations, this might be appropriate, in others, maybe not, so this is an important thing to keep in mind when deciding which sleep method to use.

In my previous post, I talked about thread deadlocks. Thread blocks, on the other hand, happen when one thread acquires a lock and holds onto it while the second thread waits for the same lock. In this case, the second thread is blocked until the first thread releases the lock. Here’s a quick example:

    private static void test() {
        final Object lock = new Object();

        Thread thread1 = new Thread(new Runnable() {
            @Override public void run() {
                synchronized (lock) {
                    System.out.println("Thread1 acquired lock");
                    try {
                        TimeUnit.MINUTES.sleep(1);
                    } catch (InterruptedException ignore) {}
                }
            }

        });
        thread1.start();

        Thread thread2 = new Thread(new Runnable() {
            @Override public void run() {
                synchronized (lock) {
                    System.out.println("Thread2 acquired lock");
                }
            }
        });
        thread2.start();
    }

Notice how Thread1 acquires the lock and does not release it for a minute. Once you run this sample, you’ll see “Thread1 acquired lock” but Thread2 does not acquire the lock for a whole minute. In that 1 minute, you can see that Thread2 is blocked in the JVM thread dump:

"Thread-2" prio=5 tid=101978000 nid=0x10a704000 waiting for monitor entry [10a703000]
   java.lang.Thread.State: BLOCKED (on object monitor)
	at BlockTest$2.run(BlockTest.java:32)
	- waiting to lock <7f3112a18> (a java.lang.Object)
	at java.lang.Thread.run(Thread.java:680)

Thread2 will eventually be unblocked, in this case, after 1 minute. So, unlike deadlocks, there’s some hope in blocks but to the end user deadlocks and blocks basically manifest themselves as unresponsive application, especially if the blocks are too frequent and long.

However, there is no excuse for programmers to confuse blocks and deadlocks as they are 2 completely different notions

What is a deadlock?

Every programmer knows what a deadlock is but for completeness sake, I’ll give a brief description. Imagine you have two threads, thread1 and thread2. Thread1 acquires lock1 and is about to acquire lock2 while thread2 acquires lock2 and is about to acquire lock1. In this case, threads are deadlocked because they are each waiting for the other thread to release the other lock and your application just hangs. Deadlocks usually occur when the order of locks is not consistent throughout the application.

Deadlock sample with regular sync blocks

Here is a a quick sample that shows the deadlock situation explained above with good old sync blocks.

    private static void test1() {
        final Object lock1 = new Object();
        final Object lock2 = new Object();

        Thread thread1 = new Thread(new Runnable() {
            @Override public void run() {
                synchronized (lock1) {
                    System.out.println("Thread1 acquired lock1");
                    try {
                        TimeUnit.MILLISECONDS.sleep(50);
                    } catch (InterruptedException ignore) {}
                    synchronized (lock2) {
                        System.out.println("Thread1 acquired lock2");
                    }
                }
            }

        });
        thread1.start();

        Thread thread2 = new Thread(new Runnable() {
            @Override public void run() {
                synchronized (lock2) {
                    System.out.println("Thread2 acquired lock2");
                    try {
                        TimeUnit.MILLISECONDS.sleep(50);
                    } catch (InterruptedException ignore) {}
                    synchronized (lock1) {
                        System.out.println("Thread2 acquired lock1");
                    }
                }
            }
        });
        thread2.start();

        // Wait a little for threads to deadlock.
        try {
            TimeUnit.MILLISECONDS.sleep(100);
        } catch (InterruptedException ignore) {}
    }

The sample prints the following and then just hangs due to deadlocked threads:

Thread1 acquired lock1
Thread2 acquired lock2

Manual Deadlock Detection

While your app is hanging like in the example above, you can get a thread dump and see the deadlocked threads. For example, on Mac, you can either do Ctrl-\ or simply use jstack and process id to get the thread dump which makes it very obvious where the deadlock is. In this example, the thread dump looks like this:

Found one Java-level deadlock:
=============================
"Thread-2":
  waiting to lock monitor 102054308 (object 7f3113800, a java.lang.Object),
  which is held by "Thread-1"
"Thread-1":
  waiting to lock monitor 1020348b8 (object 7f3113810, a java.lang.Object),
  which is held by "Thread-2"

Java stack information for the threads listed above:
===================================================
"Thread-2":
	at DeadlockTest$2.run(DeadlockTest.java:42)
	- waiting to lock <7f3113800> (a java.lang.Object)
	- locked <7f3113810> (a java.lang.Object)
	at java.lang.Thread.run(Thread.java:680)
"Thread-1":
	at DeadlockTest$1.run(DeadlockTest.java:26)
	- waiting to lock <7f3113810> (a java.lang.Object)
	- locked <7f3113800> (a java.lang.Object)
	at java.lang.Thread.run(Thread.java:680)

Found 1 deadlock.

Programmatic Deadlock Detection

In more recent JDKs, there’s programmatic deadlock detection where you can get more information about the deadlocked threads and even stacktraces leading to deadlock as explained in this stackoverflow.com entry. In our example, let’s just detect the number of blocked threads in our application:

    private static void detectDeadlock() {
        ThreadMXBean threadBean = ManagementFactory.getThreadMXBean();
        long[] threadIds = threadBean.findMonitorDeadlockedThreads();
        int deadlockedThreads = threadIds != null? threadIds.length : 0;
        System.out.println("Number of deadlocked threads: " + deadlockedThreads);
    }

If you call detectDeadlock to the end of the test1 method, you get the following nice output with the number of deadlocked threads:

Thread1 acquired lock1
Thread2 acquired lock2
Number of deadlocked threads: 2

Deadlock sample with other locks

This is all nice but what about advanced locks like ReentrantLock, does programmatic detection work in that case? Let’s use this sample:

    private static void test2() {
        final ReentrantLock lock1 = new ReentrantLock();
        final ReentrantLock lock2 = new ReentrantLock();

        Thread thread1 = new Thread(new Runnable() {
            @Override public void run() {
                try {
                    lock1.lock();
                    System.out.println("Thread1 acquired lock1");
                    try {
                        TimeUnit.MILLISECONDS.sleep(50);
                    } catch (InterruptedException ignore) {}
                    lock2.lock();
                    System.out.println("Thread1 acquired lock2");
                }
                finally {
                    lock2.unlock();
                    lock1.unlock();
                }
            }
        });
        thread1.start();

        Thread thread2 = new Thread(new Runnable() {
            @Override public void run() {
                try {
                    lock2.lock();
                    System.out.println("Thread2 acquired lock2");
                    try {
                        TimeUnit.MILLISECONDS.sleep(50);
                    } catch (InterruptedException ignore) {}
                    lock1.lock();
                    System.out.println("Thread2 acquired lock1");
                }
                finally {
                    lock1.unlock();
                    lock2.unlock();
                }
            }
        });
        thread2.start();

        // Wait a little for threads to deadlock.
        try {
            TimeUnit.MILLISECONDS.sleep(100);
        } catch (InterruptedException ignore) {}

        detectDeadlock();
    }

And the output we get is:

Thread1 acquired lock1
Thread2 acquired lock2
Number of deadlocked threads: 0

Whoa! What happened? Deadlock detection does not work with ReentrantLocks?

findMonitorDeadlockedThreads vs. findDeadlockedThreads

Well, part of the blame is mine. There are two methods in ThreadMXBean to detect deadlocks: findMonitorDeadlockedThreads and findDeadlockedThreads. I carelessly used the former but there’s a subtle difference between the two.

As JavaDocs point out, findMonitorDeadlockThreads “Finds cycles of threads that are in deadlock waiting to acquire object monitors” whereas findDeadlockedThreads “Finds cycles of threads that are in deadlock waiting to acquire object monitors or ownable synchronizers“.

What’s an ownable synchornizer? Again according to JavaDocs: “An ownable synchronizer is a synchronizer that may be exclusively owned by a thread and uses AbstractOwnableSynchronizer (or its subclass) to implement its synchronization property. ReentrantLock and ReentrantReadWriteLock are two examples of ownable synchronizers provided by the platform.”

In plain English, if you want to detect ReentrankLock or ReentrantReadWriteLock, make sure you use findMonitorThreads method. If we change our deadlock method to the following:

    private static void detectDeadlock() {
        ThreadMXBean threadBean = ManagementFactory.getThreadMXBean();
        long[] threadIds = threadBean.findDeadlockedThreads();
        int deadlockedThreads = threadIds != null? threadIds.length : 0;
        System.out.println("Number of deadlocked threads: " + deadlockedThreads);
    }

We get the following expected output with test2:

Thread1 acquired lock1
Thread2 acquired lock2
Number of deadlocked threads: 2

So, you might ask at this point, why even bother with findMonitorDeadlockedThreads? Well, findDeadlockedThreads was added in JDK6. So, if your project is using JDK5 still, you need to use findMonitorDeadlockedThreads and hopefully this post will remind you the caveats of using findMonitorDeadlockedThreads.

Typically, a client would connect to a JMX enabled server using a URL similar to this:

service:jmx:rmi:///jndi/rmi://host:port1/jmxrmi

This basically tells the client to connect to a JMX server on the specified host and port over RMI. At least, that’s what you’d expect right? Well, the reality is a little different. When the client connects to host over that port, the JMX server responds with something like “Ok, I got your connect request on port1, now you can connect on port2” and accordingly to this blog post, port2 is chosen randomly.

Why is this problematic? Imagine trying to administer your JMX enabled server through a firewall where a predetermined number of ports are open. You can see how random port2 makes life quite difficult. So what can you do? JMXMP comes to rescue. There’s not much info on JMXMP out there but according to JavaDocs:

The JMX Messaging Protocol (JMXMP) connector is a configuration of the generic connector where the transport protocol is based on TCP and the object wrapping is native Java serialization. Security is more advanced than for the RMI connector. Security is based on the Java Secure Socket Extension (JSSE), the Java Authentication and Authorization Service (JAAS), and the Simple Authentication and Security Layer (SASL).

It looks like JMXMP is not only better in terms of getting rid of random ports but it has better security. The only caveat is that JMXMP connector is optional (i.e. JDK does not include it by default). All you have to do is Google for the download page for jmxremote_optional.jar from Oracle or if you’re a Maven user, simply add the following to your pom.xml.

Once you have jmxremote_optional.jar in classpath of your client and server, you can connect your client to your JMX server using a URL like this with your own host and port:

service:jmx:jmxmp://host:port

The URL looks much cleaner than RMI, the port you specify is the only port you need to worry about and it’s more secure. I have no idea why JMXMP is not the default connector in JMX.

If you want to run the application yourself, here’s the zip file with the Java classes I mentioned in the video.

As some of you might know, Java’s Socket class has a reuseAddress property that is supposed to do the following according to JavaDocs:

When a TCP connection is closed the connection may remain in a timeout state for a period of time after the connection is closed (typically known as the TIME_WAIT state or 2MSL wait state). For applications using a well known socket address or port it may not be possible to bind a socket to the required SocketAddress if there is a connection in the timeout state involving the socket address or port. Enabling SO_REUSEADDR prior to binding the socket using bind(SocketAddress) allows the socket to be bound even though a previous connection is in a timeout state.

In more simpler terms, when reuseAddress is enabled on a socket on the server, the server socket can bind to that port even if there’s a client connection in a TIME_WAIT state on that port. This comes up when the client is connected to the server, and the server goes down for whatever reason. In that scenario, the connection gets to a TIME_WAIT state, because it’s not an orderly shutdown, and when you restart the server, the server cannot bind to that port until after a OS specific timeout. This means server restarts can take up a while, and hence the need of a reuseAddress property.

In LCDS, there’s also a “reuse-address-enabled” property to take advantage of this Socket property but we recently realized that it didn’t work as expected on Linux. So, even when reuseAddress property was set to true, server restarts would end up with bind exceptions on Linux and we didn’t know why until a coworker, Alex Glosband, pointed out the following from the Linux man page for socket:

Linux will only allow port re-use with the SO_REUSEADDR option when this option was set both in the previous program that performed a bind() to the port and in the program that wants to re-use the port. This differs from some implementations (e.g., FreeBSD) where only the later program needs to set the SO_REUSEADDR option. Typically this difference is invisible, since, for example, a server program is designed to always set this option.

In LCDS, when “reuse-address-enabled” property was true, LCDS tried to bind to the port with reuse-address=false first. When it failed, it would log that it couldn’t bind to the port, then, it would set reuse-address=true, and bind again. This works fine on Windows or Mac, but not on Linux, because Linux expects that all binds to a port have reuse-address=true for address reuse to work. In the end, the fix was trivial, we changed to logic to use reuse-address=true in the first bind try on Linux, and not worry about logging about the first bind failure.