Bill Bejeck

in this post, we introduce how to use .NET Kafka clients along with the Task Parallel Library to build a robust, high-throughput event streaming application.

Understanding the role of a partitioner in the Kafka producer is important to understand to build effective Kafka applications.

Let's see how to optimize Kafka Streams applications.

Java 8's debut included a neat concurrency too, the CompletableFuture class. Here's an awesome look at CompletableFutures, with a glance at combining and composing tasks.

Learn how you can use partially applied functions in Java 8

This is a short post about a method I recently discovered in Guava. The Issue I had a situation at work where I was working with objects structured something like this: public class Outer { String outerId; List innerList; ....... } public class Inner { String innerId; Date timestamp; } public class Merged { String outerId; String innerId; Date timestamp; } My task was flatten a list Outer objects (along with the list of Inner objects) into a list of Merged objects. Since I’m working with Java 7, using streams is not an option. The First Solution Instead I turn to the FluentIterable class from Guava. My first instinct is to go with the FluentIterable.transform method (which is essentially a map function): List originalList = getListOfObjects(); Function> flattenFunction //Details left out for clarity //returns an Iterable of Lists! Iterable> mergedObjects = FluentIterable.from(originalList).tranform(flattenFunction); But I really want a single collection of Merged objects, not an iterable of lists! The missing ingredient here is a flatMap function. Since I’m not using Scala, Clojure or Java 8, I feel that I’m out of luck. A Better Solution I decide to take a closer look at the FluentIterable class and I discover the FluentIterable.transformAndConcat method. The transformAndConcat method applies a function to each element of the fluent iterable and appends the results into a single iterable instance. I have my flatMap function in Guava! Now my solution looks like this: List originalList = getListOfObjects(); Function> flattenFunction //Details left out for clarity Iterable mergedObjects = FluentIterable.from(originalList).transformAndConcat(flattenFunction); Conclusion While this is a very short post, it goes to show how useful the Guava library is and how functional programming concepts can make our code more concise.

The Java 8 lambda syntax and functional interfaces have been a productivity boost for Java developers. But there is one drawback to functional interfaces. None of the them as currently defined in Java 8 declare any checked exceptions. This leaves the developer at odds on how best to handle checked exceptions. This post will present one option for handling checked exceptions in functional interfaces. We will use use the Function in our example, but the pattern should apply to any of the functional interfaces. Example of a Function with a Checked Exception Here’s an example from a recent side-project using a Function to open a directory in Lucene. As expected, opening a directory for writing/searching throws an IOException: Create a Lucene Directory private Function createDirectory = path -> { try{ return FSDirectory.open(path); }catch (IOException e){ throw new RuntimeException(e); } }; While this example works, it feels a bit awkward with the try/catch block. It’s adding to the boilerplate that functional interfaces are trying to reduce. Also, we are just re-throwing the exception up the call stack. If this how we are going to handle exceptions, is there something we can do to make our code a little bit cleaner? A Proposed Solution The solution is straight forward. We are going to extend the Function interface and add a method called throwsApply. The throwsApply method declares a throws clause of type Exception. Then we override the applymethod as a default method to handle the call to throwsApply in a try/catch block. Any exceptions caught are re-thrown as RuntimeExceptions Extending the Function Interface @FunctionalInterface public interface ThrowingFunction extends Function { @Override default R apply(T t){ try{ return applyThrows(t); }catch (Exception e){ throw new RuntimeException(e); } } R applyThrows(T t) throws Exception; } Here we are doing exactly what we did in the previous example, but from within a functional interface. Now we can rework our previous example to this: (made even more concise by using a method handle) using Function that handles checked Exceptions private ThrowingFunction createDirectory = FSDirectory::open; Composing Functions that have Checked Exceptions Now we have another issue to tackle. How do we compose two or more functions involving checked exceptions? The solution is to create two new default methods. We create andThen and compose methods allowing us to compose ThrowingFunction objects into one. (These methods have the same name as the default methods in the Function interface for consistency.) New Default method andThen default ThrowingFunction andThen(ThrowingFunction after){ Objects.requireNonNull(after); try{ return (T t) -> after.apply(apply(t)); }catch (Exception e){ throw new RuntimeException(e); } } default ThrowingFunction compose(ThrowingFunction before) { Objects.requireNonNull(before); try { return (V v) -> apply(before.apply(v)); }catch (Exception e){ throw new RuntimeException(e); } } The code for theandThen and compose is the same found in the original Function interface. We’ve just added try/catch blocks and re-throw any Exception as a RuntimeException. Now we are handling exceptions in the same manner as before with the added benefit of our code being a little more concise. Caveats This approach is not without its drawbacks. Brian Goetz spoke about this in his blog post ‘Exception transparency in Java’. Also, when composing functions we must use the type of ThrowingFunction for all parts. This is regardless if some of the functions don’t throw any checked exceptions. There is one exception, the last function added could be of type Function. Conclusion The purpose of this post is not to say this the best approach for handling checked exceptions, but to present one option. In a future post we will look at other ways of handling checked exceptions in functional interfaces. Resources Source for post Stackoverflow topic Functional Programming in Java presents good coverage of how to handle checked exceptions in functional interfaces. Java 8 Lambdas Throwing Checked Exceptions From Lambdas

This Post is a continuation of my series on Google Guava, this time covering Guava Cache. Guava Cache offers more flexibility and power than either a HashMap or ConcurrentHashMap, but is not as heavy as using EHCache or Memcached (or robust for that matter, as Guava Cache operates solely in memory). The Cache interface has methods you would expect to see like ‘get’, and ‘invalidate’. A method you won’t find is ‘put’, because Guava Cache is ‘self-populating’, values that aren’t present when requested are fetched or calculated, then stored. This means a ‘get’ call will never return null. In all fairness, the previous statement is not %100 accurate. There is another method ‘asMap’ that exposes the entries in the cache as a thread safe map. Using ‘asMap’ will result in not having any of the self loading operations performed, so calls to ‘get’ will return null if the value is not present (What fun is that?). Although this is a post about Guava Cache, I am going to spend the bulk of the time talking about CacheLoader and CacheBuilder. CacheLoader specifies how to load values, and CacheBuilder is used to set the desired features and actually build the cache. CacheLoader CacheLoader is an abstract class that specifies how to calculate or load values, if not present. There are two ways to create an instance of a CacheLoader: Extend the CacheLoader class Use the static factory method CacheLoader.from If you extend CacheLoader you need to override the V load(K key) method, instructing how to generate the value for a given key. Using the static CacheLoader.from method you build a CacheLoader either by supplying a Function or Supplier interface. When supplying a Function object, the Function is applied to the key to calculate or retrieve the results. Using a Supplier interface the value is obtained independent of the key. CacheBuilder The CacheBuilder is used to construct cache instances. It uses the fluent style of building and gives you the option of setting the following properties on the cache: Cache Size limit (removals use a LRU algorithm) Wrapping keys in WeakReferences (Strong references used by default for keys) Wrapping values in either WeakReferences or SoftReferences (Strong references used by default) Time to expire entires after last access Time based expiration of entries after being written or updated Setting a RemovalListener that can recieve events once an entry is removed from the cache Concurrency Level of the cache (defaults to 4) The concurrency level option is used to partition the table internally such that updates can occur without contention. The ideal setting would be the maximum number of threads that could potentially access the cache at one time. Here is an example of a possible usage scenario for Guava Cache. public class PersonSearchServiceImpl implements SearchService> { public PersonSearchServiceImpl(SampleLuceneSearcher luceneSearcher, SampleDBService dbService) { this.luceneSearcher = luceneSearcher; this.dbService = dbService; buildCache(); } @Override public List search(String query) throws Exception { return cache.get(query); } private void buildCache() { cache = CacheBuilder.newBuilder().expireAfterWrite(10, TimeUnit.MINUTES) .maximumSize(1000) .build(new CacheLoader>() { @Override public List load(String queryKey) throws Exception { List ids = luceneSearcher.search(queryKey); return dbService.getPersonsById(ids); } }); } } In this example, I am setting the cache entries to expire after 10 minutes of being written or updated in the cache, with a maximum amount of 1,000 entires. Note the usage of CacheLoader on line 15. RemovalListener The RemovalListener will receive notification of an item being removed from the cache. These notifications could be from manual invalidations or from a automatic one due to time expiration or garbage collection. The RemovalListener parameters can be set to listen for specific type. To receive notifications for any key or value set them to use Object. It should be noted here that a RemovalListener will receive a RemovalNotification object that implements the Map.Entry interface. The key or value could be null if either has already been garbage collected. Also the key and value object will be strong references, regardless of the type of references used by the cache. CacheStats There is also a very useful class CacheStats that can be retrieved via a call to Cache.stats(). The CacheStats object can give insight into the effectiveness and performance of your cache by providing statistics such as: hit count miss count total load timme total requests CacheStats provides many other counts in addition to the ones listed above. Conclusion The Guava Cache presents some very compelling functionality. The decision to use a Guava Cache really comes down to the tradeoff between memory availability/usage versus increases in performance. I have added a unit test CacheTest demonstrating the usages discussed here. As alway comments and suggestions are welcomed. Thanks for your time. Resources Guava Project Home Cache API Source Code for blog series From http://codingjunkie.net/google-guava-cache/