Random Tech Rants

UnicastProcessor sample

2019-07-18T12:55:00.000-04:00

UnicastProcessor

UnicastProcessor hotSource = UnicastProcessor.create()
hotSource.subscribe(System.out::println)
hotSource.onNext("foo");
hotSource.onNext("bar");
. . .

Optimized character search in String

2016-12-01T18:13:00.000-05:00

Problem:
You have a large string (string A) and another string (string B) and you want to find out if all characters of string B are present in string A. The problem is not complex through a basic loop and ‘contains’ call, but could be made more optimal by splitting both strings. Why splitting is important? By splitting string A you creating a potential to discover that all character in string B are present in split from String A and you can break off the loop and return true. If not you concatenate first split with the second one and repeat. The bottom line is that you only have to look through the whole string if you can’t find a match until the last split. Splitting string B is also important since if a split from string B was matched to a string A, you no longer have to include it in your matching logic.

public class CharSearchInString {

public static void main(String[] args) {

String[] driverStrings = "java.util.concurrent.atomic.AtomicInteger".split("(?<=\\G.{4})");

String[] comparableStrings = "avrutcimIr".split("(?<=\\G.{3})");

// should return true as all characters in 'comparableStrings' are

// present in 'driverStrings'

System.out.println(performCompare(driverStrings, comparableStrings));

}

/**

private static boolean performCompare(String[] driverStrings, String[] comparableStrings) {

int csCount = 0;

String driverString = "";

for (int dsCount = 0; dsCount < driverStrings.length;) {

innerLoop:

for (; csCount < comparableStrings.length;) {

String comparableString = comparableStrings[csCount];

if (compares(driverString, comparableString)) {

csCount++;

} else {

driverString += driverStrings[dsCount];

dsCount++;

}

break innerLoop;

}

if (csCount == comparableStrings.length) {

break;

}

// will return true if all strings in 'comparableStrings' were matched

// to at least one string in 'driverStrings'

return csCount == comparableStrings.length;

}

/**

private static boolean compares(String a, String b) {

byte[] bArray = b.getBytes();

for (int i = 0; i < bArray.length; i++) {

byte x = bArray[i];

if (!a.contains((char) x + "")) {

return false;

}

return true;

}

Set static final via java Reflection

2016-06-09T12:09:00.000-04:00

public static void setFinalField(Field field, Object newValue) throws Exception {

field.setAccessible(true);

Field modifiersField = Field.class.getDeclaredField("modifiers");

modifiersField.setAccessible(true);

modifiersField.setInt(field, field.getModifiers() & ~Modifier.FINAL);

field.set(null, newValue);

}

Getting stack trace without raising exception

2015-03-14T09:30:00.001-04:00

Thread.currentThread().getStackTrace()

Make Java 8 Lambdas Serializable (type intersection)

2015-02-28T11:28:00.001-05:00

Java 8 introduces type intersection allowing object to be cast to an intersection of types

public class TypeIntersectionSample {

public static void main(String... arg) {

Serializable bser = toSerializable(Integer::sum);

}

public static <T extends BinaryOperator<Integer> & Serializable> T toSerializable(T bo) {

return (T) bo;

}

ThreadLocal with Java 8

2015-02-09T15:49:00.002-05:00

There is a new factory method in ThreadLocal allowing you to create and set initial value all at he same time via lambda expression.

ThreadLocal tl = ThreadLocal.withInitial(() -> "Hello");

The rest is the same as before with regard to set(..)/get().

Accessing JVM's memory info

2015-01-23T15:36:00.003-05:00

Simply use MemoryMXBean

MemoryMXBean memoryBean = ManagementFactory.getMemoryMXBean();
MemoryUsage heapUsage = memoryBean.getHeapMemoryUsage();
MemoryUsage nonHeapUsage = memoryBean.getNonHeapMemoryUsage();

Then simply interrogate instance of MemoryUsage

long usedMemory = heapUsage.getUsed();

long maxMemory = heapUsage.getMax();

long initMemory = heapUsage.getInit();

long maxMemory = heapUsage.getMax();

The same goes for non heap

Messos (Cliffs Notes)

2015-01-22T11:44:00.001-05:00

What is it?
Mesos is an independent cluster platform which hosts cluster computing frameworks.
Giving the fact that cluster platform's main responsibility to manage cluster resources a good example of similar architecture in Hadoop would be YARN and Tez, where YARN is native Hadoop cluster platform with Tez being domain-specific (domain being DAG) cluster computing framework hosted by YARN. This allows not only multi-framework hosting but also multi-instances and/or versions of the same framework hosted in the same shared cluster.

Architecture
Mesos - Consists of Master and Slave nodes with allocation modules hosted by Master
Framework - Consists of Scheduler and Executor processes launched on every node (to execute framework's tasks)

Scheduling and data locality
Mesos introduces the concept of decentralized scheduler. The arguments for that are:

the need for expressive API to support all the frameworks (known and un-known)
new frameworks may already come with its own scheduling mechanism, so refactoring it into centralized would not be feasible

So Mesos delegates scheduling to the frameworks themselves via resource offers. Mesos offers resources to frameworks based on organizational policies (e.g., fair, priority etc), while the framework itself decides to accept/reject or how much of resources to accept.

Slave reports resource availability to the Master ->
Master(4 cpu and 8Gb on this slave) via allocation policy module ->
Framework's scheduler (takes 2 CPU, 2GB mem, on this slave or rehect) ->
Master(OK, and launches the tasks on such slave)

Unused resources are offered to other frameworks and the overall process repeats upon completion of tasks.

This is one of the mechanism frameworks utilize to achieve data locality - by accepting and/or rejecting resource offers from slaves based on data proximity.

So the main design philosophy for Mesos - define a minimal interface that enables efficient resource sharing across frameworks, and otherwise push control of task scheduling and execution to the frameworks.

Allocation module
Based on assumption that task have a short life span which means it can reallocate resources as quick as tasks complete. However in a cases of long-running tasks Mesos can revoke (kill) tasks by killing task itself and/or killing executor if task does not respond. It's all in implementation detail of allocation module policies.
While MR based frameworks can easily live with re-allocated/killed tasks, MPI-style frameworks can't, so allocation modules can expose guaranteed allocation to each framework - basically a set of resources permanently taken by the framework.

Since resource acceptance is based on accept or reject mechanism, rejection in itself may introduce an unnecessary chatter especially for cases where certain slaves would always reject a particular scheduling request. Mesos allows such chatter to be further reduced by providing allocation filters to the master essentially allowing resource offer not to go out to a particular slave if it does not pass a filter criteria

Resource offers will be rescinded if a framework does not respond timely to resource offer, thus allowing such resources to be re-offered to other frameworks.

Performance Isolation
Performance isolation achieved via native mechanisms (e.g., Linux containers, etc.)

Master fault tolerance (hot stand-by)
Master has few standby nodes. In a case of master failure ZooKeeper will select a new master from the list. Since all (hot and standby) masters receive events from the cluster, a new master will be able to reconstruct its latest state from these events.

Opinion:
Mesos may have its benefits in heterogeneous clusters often seen in small/medium sized businesses where maintaining a domain specific cluster is not cost efficient due to low load and high idle scenarios, so a single cluster platform can host multiple cluster computing frameworks sharing resources between them, thus limiting the idle time.
However, in large scale high load data clusters (e.g., Hadoop) having a native cluster platform such as YARN is more beneficial based on a simple premise that specialized solutions are generally better then generalized.
Having said that, YARN is perfectly capable of hosting cluster computation frameworks which may have nothing to do with Hadoop all together, thus allowing Hadoop/YARN cluster to be used as heterogeneous cluster compute platform.

Missrepresentation:
In Mesos Architecture white paper (http://mesos.berkeley.edu/mesos_tech_report.pdf). In Page 2 Section 2 it talks about an example of what needs to happen to data sitting in Hadoop after MR processing if such data needs to be integrated with MPI. It talks about requirement for spinning up a new MPI cluster and moving terrabytes of data from Hadoop cluster while making the argument that MPI could be just another compute framework hosted in the same cluster.
That is actually not true and these capabilities existed in Hadoop even before YARN/Tez, since MPI concepts could always be implemented as part of your MR job (although I'd admit it was not simple). However with YARN and Tez it is relatively simple. In fact Tez is being more flexible when it comes to this sort of requirements since it is a low-level DAG framework comprised of simple Vertexes "processors" (filters) connected by Edges (pipes) as in standard pipes-and-filters architecture.

Java 8 - "The target type of this expression must be a functional interface"

2015-01-13T14:01:00.001-05:00

When learning new constructs of Java 8 you quickly learn about method references (or functional references). The similarity is also often drawn to object references. This may lead to one of these mistakes:

Object function = File::isHidden;

The above will result in compile error "The target type of this expression must be a functional interface".
The correct way is to use the appropriate functional interface:

Predicate<File> function = File::isHidden;

Where the class is loaded from?

2014-12-13T17:15:00.002-05:00

Variety of exceptions could be categorized as class loading exceptions, since their root cause most often leads one to discover issues with the classpath. Some of these are NoClassDefFoundError, LinkageError, NoSuchMethodError etc. Debugging these issues is greatly simplified when you know where a class is loaded from (i.e., which JAR or directory).
The code below will extract the URL for the location of the source for a particular class (Foo in this case)

URL fooClassSourceUrl = Foo.class.getProtectionDomain().getCodeSource().getLocation()

Object instance without constructor invocation

2014-12-13T15:47:00.000-05:00

There are times when you need to create an instance of an object without invoking its constructor. This is typical in testing (e.g. mock) or integrating where you may need to comply with method that may perform a null check while downstream such object may not be used. One of those un-documented classes in Java API (sun.misc.Unsafe) could help.

1. Create an instance of sun.misc.Unsafe:
import java.lang.reflect.Constructor;

import sun.misc.Unsafe;

. . .

public static Unsafe getUnsafe() throws Exception {

Constructor ctr = Unsafe.class.getDeclaredConstructor();

ctr.setAccessible(true);

return ctr.newInstance();

}

2. Create an instance of an object without invoking its constructor:

Foo foo = (Foo) getUnsafe().allocateInstance(Foo.class);

Caution: you can invoke any method on such instance as if it was created using a constructor. However, realize the effects of this "back-door" where invocation of any method on such uninitialized instance could and most likely would lead to various exceptions.

Tez container memory allocations

2014-12-11T09:04:00.005-05:00

As a YARN application Tez must allocate YARN containers for Application Master and Tasks.
The relevant properties and their default values are:

tez.am.resource.memory.mb - 1536 (MB)
tez.task.resource.memory.mb - 1024 (MB)

When setting, ensure that AM memory + Task memory fits in the overall capacity of the cluster (see YARN total memory)

YARN total memory

2014-12-11T08:51:00.002-05:00

Shown as Memory Total in YARN Resource Manager UI

To change it, modify yarn.nodemanager.resource.memory-mb property.
Could me modified with Ambari.

Scala Named parameters

2011-07-20T12:30:00.000-04:00

Named parameters is one of the features of Scala which allows you to define a class which has a constructor that defines default values for the named parameters which you can later override.

For example:

class Person(name: String = "Bill", age: Int = 25) {
      override def toString = name + ":" + age
}

First, this allows you to create an instance of this class without passing any arguments

println(new Person())

and the output will be:

Bill:25

This is because the instance is created with the default values specified in the class definition.

You can override any of these values by using the name of the argument and specifying its new value

For example:

println(new Person(age=35))

will result in:

Bill:35

and

println(new Person(age=55, name="John"))

will result in:

John:55

And of course you can still do this:

println(new Person("Chris", 23))

Implementing Spring's FactoryBean in Scala

2011-07-06T11:01:00.000-04:00

Spring Framework provides many different extension points. One of them is FactoryBean
The details about FactoryBeans are described here. All I'll show is how to implement one in Scala, by extending on the previous post Scala Function as Spring Bean (Spring and Scala)
Let's say creating an instance of the Scala function requires some logic that could not or should not be handled by a simple constructor invocation. This would be a perfect case for a FactoryBean and we can easily implement one in Scala

So to create our PrintFunction via FactoryBean we can implement one in Java or in Scala. Below is the Scala implementation of Spring's FactoryBean which creates and instance of PrintFunction while also printing a message stating that it is creating a PrintFunction.

class ScalaFactoryBean extends FactoryBean[PrintFunction] {
  
  val myFunction = new PrintFunction()

  def getObject(): PrintFunction = {  
    println("Creating 'PrintFunction'")
 return myFunction
  }

  def getObjectType(): Class[PrintFunction] = { 
    classOf[PrintFunction]
  }

  def isSingleton(): Boolean = { 
    true 
  }
}

. . .and its configuration:

</bean>

From this point on it is pure Spring

public static void main(String[] args) {
    ApplicationContext context = new ClassPathXmlApplicationContext("scala-config.xml", SpringDemo.class);
  
    SimpleSpringBean functionViaFactoryBean = context.getBean("functionViaFactoryBean", SimpleSpringBean.class);
    functionViaFactoryBean.printMessage("Hello Spring-Scala");
}

For more info check out the source code here: https://github.com/olegz/scala-spring-samples/tree/master/src/test/scala/olegz/scala/spring

Scala Function as Spring Bean (Spring and Scala)

2011-07-06T09:51:00.000-04:00

With Scala gaining popularity a lot of developers are now wondering how Scala can integrate with existing and popular JVM-based frameworks.
One of the questions I've been asked recently if Scala Functions could be used as Spring Beans in the typical Dependency Injection model provided by Spring Framework. In other words can I inject Scala function into a Spring configured Java bean? The answer is not only 'Yes' but it is actually very simple, since Scala itself is very nicely integrated with Java and Scala Function is just a class.

Let's say we have a Scala function called PrintFunction

class PrintFunction extends Function1[String, Unit]{
   def apply(in: String) = println("From Scala function: " + in)
}

It is worth pointing out that Scala defines a group of traits from scala.Function0 to scala.Function9, allowing you to define functions with 0 to 9 parameters so you can easily integrate them with other classes. In our case we are using Function1 since we only passing one parameter. However we can clearly see that there is a second type argument in our definition of PrintFunction (e.g., Function1[String, Unit]). That is the return value and in our case it is Unit which would be the equivalent of Java 'void'.

So, to bootstrap this function as a Spring Bean all we need is to define it as Spring Bean

Now it is just another Spring Bean which could be injected into any other bean that has PrintFunction property (see below).

public class SimpleSpringBean {

    private PrintFunction function;

    public void setFunction(PrintFunction function) {
        this.function = function;
    }
        
    public void printMessage(String message) {
        function.apply(message);
    }
}

. . . and its configuration:

</bean>

Now all we need is start Spring Application Context and call printMessage(String) method of SimpleSpringBean. This method will invoke Function's apply(String) method, or we can simply get the function itself and call its apply(String) method.

public static void main(String[] args) {
    ApplicationContext context = new ClassPathXmlApplicationContext("scala-config.xml", SpringDemo.class);
    SimpleSpringBean bean = context.getBean(SimpleSpringBean.class);
    bean.printMessage("Hello Spring-Scala");
}

You should see the following output:

From Scala function: Hello Spring-Scala

For more details check out this sources here https://github.com/olegz/scala-spring-samples/tree/master/src/test/scala/olegz/scala/spring

Spring Integration and Channels

2011-07-04T16:46:00.000-04:00

Spring Integration is a POJO-based light weight, embeddable messaging framework with a loosely coupled programming model aimed to simplify integration of heterogeneous systems based on Enterprise Integration Patterns and without requiring a separate ESB-like engine or proprietary development and deployment environment.

One of the fundamental differences between Spring Integration and other EIP-like frameworks and products is that we treat EIP as specification and follow it to the letter.
At the very core of the EIP is one fundamental and probably the most important pattern of all; Pipes-and-Filters which describes the communication model between Message producers and Message consumers where instead of exchanging Messages directly producers and consumers communicate through the pipe (i.e., channel)

Pipe is a core component of EIP for 2 main reasons:

1. Pipes give you logical and physical decoupling. Because of that components on either side of the pipe are completely un-aware of one another. You can change producing or consuming component without affecting the flow (e.g., consumer could be a mock while in the early developing stages and than switched to a real component later on)

2. Pipes in collaboration with Message Dispatcher and Polling Consumer are responsible for handling Message exchange protocol (e.g., point-to-point vs. pub-sub, sync or async, etc.).
Let's look at the example:

producerA -> channelA -> consumerA

In the above diagram producerA sends Message to a consumerA via channelA.
As you can see neither producer nor consumer are aware of one another. They are physically and logically decoupled. We can change producer without affecting the consumer and vice versa.

Message exchange protocol is being maintained by the channel? Is it point-to-point? pub-sub? sync or async? Well the diagram above does not show it and in Spring Integration the default channel implementation is synchronous point-to-point channel. However by simply changing the type of channel, one can completely change the Message exchange protocol without affecting producer or consumer.

JMX Connectivity through Firewall

2008-12-08T14:19:00.002-05:00

Recently I’ve been asked to help out a customer who was having issues with JMX connectivity to Spring Source dmServer through the firewall. However, one thing I want to point out right up front is that the issue is rather generic and has nothing to do with dmServer. It is really about understanding JMX, RMI and proper configuration. But I will use dmServer and its configuration as an example.
Here is the sample JMX configuration options provided in the dmServer startup script:

-Dcom.sun.management.jmxremote.port=${jmxPort} \
-Dcom.sun.management.jmxremote.authenticate \
-Dcom.sun.management.jmxremote.password.file=${jmxUsersPath} \
-Djavax.net.ssl.keyStore=${keystorePath} \
-Djavax.net.ssl.keyStorePassword=${keystorePassword} \
-Dcom.sun.management.jmxremote.ssl=true \
-Dcom.sun.management.jmxremote.ssl.need.client.auth=false”

This will enable JMX agent (MBean Server) when you start dmServer. Once started you can now monitor your process via JMX-compliant tool such as jconsole. Connectivity could be local or remote.
The above configuration seem to provide everything we need to access this process through the firewall, since com.sun.management.jmxremote.port is obviously the port that we need to open in the firewall. However there is a caveat.
Once connected to JMXRegistry running on the port specified by com.sun.management.jmxremote.port property, the actual objects are served by RMIServer which is running on different port. Unfortunately this port is chosen randomly by default instance of JMX Agent and there is no –D option to specify it. Obviously going through the firewall would require opening up two ports and with random port it presents a delicate issue.
Fortunately it is easily solvable by writing a custom Java Agent where you can programmatically specify each port and externalize it through custom properties (I am attaching sample code).
More info here: http://java.sun.com/javase/6/docs/technotes/guides/management/agent.html
In the nutshell, the custom agent will take the port value provided by the com.sun.management.jmxremote.port property and will create a second port (RMIServer port) by incrementing it by 1. (in our case the port specified is 44444 which makes RMIServer port 44445)
Once such agent is in place (JAR) and the appropriate ports are open in the firewall all you need is modify the startup script to include –javaagent option providing the JAR.

. . . . .
$JAVA_HOME/bin/java \
-javaagent:/Users/olegzhurakousky/../../../dmserver.jmx-0.0.1-SNAPSHOT.jar 
$JAVA_OPTS \
. . . . .

Well, that really only solved one half of the problem, since by default RMI stubs sent to the client contain the server’s private address instead of the public

Just look at this tcpdump fragment while monitoring the client’s access (jconsole running on the local network):

. . . . . . .
09:41:23.778663 IP 72.234.14.89.44444 > 192.168.1.156.52926: . ack 71 win 65535
09:41:23.779958 IP 192.168.1.152.44444 > 72.234.14.89.52926: P 20:251(231) ack 71 win 65535
09:41:23.780456 IP 72.234.14.89.44444 > 192.168.1.156.52926: P 20:251(231) ack 71 win 65535
09:41:23.796075 IP 192.168.1.156.37861 > 192.168.1.152.44445: S 1334070579:1334070579(0) win 5840
09:41:23.796328 IP 192.168.1.152.44445 > 192.168.1.156.37861: S 1760846938:1760846938(0) ack 1334070580 win 65535
. . . . . . .

You can clearly see that 192.168.1.156 (client i.e., jconsole) is attempting to connect directly to 192.168.1.152 (server) instead of 72.234.14.89 which is a public IP, although the JMX URL is:
service:jmx:rmi://72.234.14.89:44445/jndi/rmi://72.234.14.89:44444/jmxrmi

If I was behind the firewall I would obviously had problems connecting to 192.168.1.152
Fortunately, this one is easy to fix. All you need is to provide additional option on the server side (java.rmi.server.hostname) and add it to the script This option represents the host name string that should be associated with remote stubs for locally created remote objects, in order to allow clients to invoke methods on the remote object:

. . . . . . .
JMX_OPTS=” \
$JMX_OPTS \
-Dcom.sun.management.jmxremote.port=${jmxPort} \
-Djava.rmi.server.hostname=72.234.14.89 \
. . . . . . .

That is all .
Start jconsole: ./jconsole.sh service:jmx:rmi://:/jndi/rmi://:/jmxrmi
Once you modify the script and start the dmServer you should see output similar to this:

. . . . . .
oleg-2:bin olegzhurakousky$ ./startup.sh
com.springsource.rmiregistry.port:44444
com.springsource.rmiserver.port:44445
Getting the platform’s MBean Server
Local Connection URL: service:jmx:rmi://oleg-2.local:44445/jndi/rmi://oleg-2.local:44444/jmxrmi
Public Connection URL: service:jmx:rmi://72.234.14.89:44445/jndi/rmi://72.234.14.89:44444/jmxrmi
Creating RMI connector server
[2009-02-26 18:53:34.031] main Server starting.
[2009-02-26 18:53:35.943] main OSGi telnet console available on port 2401.
[2009-02-26 18:53:41.558] main Boot subsystems installed.