Exception Handling in Java Streams

Java 8 introduced the Java Streams API to allow functional-style operations on streams of data. Streams represent a sequence of elements and support different kinds of methods like filter, map, reduce and so on. Streams can be created from various data sources like collections, arrays, files, regular expressions etc.

While streams provide a powerful and concise way to manipulate data, exception handling when using streams requires some care. We will look at how exceptions are handled in streams and best practices for handling them properly in this article.

Exceptions in Intermediate and Terminal Operations

Operations on streams can be divided into intermediate and terminal operations.

Intermediate operations like filter, map, sorted etc. return another stream. They are lazy, meaning they don’t process the data until a terminal operation is called.

Terminal operations like forEach, count, collect etc. return a non-stream result like a primitive, collection or void. They process the stream pipeline and return a result.

Exceptions thrown by intermediate operations are not handled or reported immediately. They are encapsulated in the returned stream. Only when a terminal operation is invoked do these exceptions surface.

For example:

Stream<String> stream = Stream.of("a", "b", "c");
Stream<String> filteredStream = stream.filter(s -> {
  throw new RuntimeException("Exception from filter"); 
});

filteredStream.forEach(s -> {
  System.out.println(s); 
});

Here, the filter operation throws an exception but the exception does not occur immediately. Only when the terminal forEach operation is called does the exception surface. In other words, If you don’t call any terminal operation, stream does not throw exception.

You can test it with the below code.

Stream<String> stream = Stream.of("a", "b", "c");
Stream<String> filteredStream = stream.filter(s -> {
  throw new RuntimeException("Exception from filter"); 
});

On the other hand, exceptions from terminal operations are handled then and there:

Stream<String> stream = Stream.of("a", "b", "c");

stream.forEach(s -> {
  throw new RuntimeException("Exception from forEach");
});
System.out.println("Done"); // Not printed due to exception

Here, the exception thrown in the terminal forEach operation surfaces immediately.

Handling Checked Exceptions

Intermediate operations cannot throw checked exceptions directly. Because, they need to be declared or caught. So these exceptions are wrapped in an Unchecked exception by the streams API.

For example:

Stream<String> stream = Files.lines(path); //IOException possibility

stream.map(s -> {
  // IOException wrapped in Unchecked exception
  throw new RuntimeException(new IOException()); 
})
.forEach(s -> {
  System.out.println(s);
});

The IOException is wrapped in a RuntimeException and thrown only when the terminal operation is invoked.

To handle checked exceptions properly, you need to handle them at the appropriate place in the stream pipeline. For example:

try {
  Stream<String> stream = Files.lines(path);
  stream.map(s -> {
    // do something
  })
  .forEach(s -> {
    System.out.println(s);
  });
} catch (IOException e) {
  // handle IOException
}

Here, we place the stream initialization code in the try block and handle the checked IOException in the catch block. This is cleaner than handling the wrapped exception in the terminal operation.

Handling Exceptions from map and flatMap

The intermediate map and flatMap operations can also throw exceptions which are encapsulated in the returned stream. To handle these properly, the exception needs to be caught from the lambda expression itself.

For example:

Stream<String> stream = Stream.of("a", "b", "c");

stream.map(s -> {
  try {
    // logic that throws exception
  } catch (Exception e) {
    // handle exception 
  }
  return s; 
})
.forEach(s -> {
  System.out.println(s); 
});

Here, we catch the exception inside the map lambda expression instead of the terminal operation. This way, we handle it at the source instead of later in the pipeline.

The same applies for flatMap which maps each input to a stream.

Use Collectors.partitioningBy for Exceptional Cases

Collectors.partitioningBy is a useful collector. By using it, you can partition the stream elements based on whether they pass a condition or throw an exception.

List<String> data = Arrays.asList("1", "2", "three", "4", "five");

Map<Boolean, List<String>> partitioned = data.stream()
    .collect(Collectors.partitioningBy(
        str -> {
            try {
                Integer.parseInt(str);
                return true; // No exception, pass the condition
            } catch (NumberFormatException e) {
                return false; // Exception occurred
            }
        }
    ));
List<String> validNumbers = partitioned.get(true);
List<String> invalidInputs = partitioned.get(false);
System.out.println("Valid Numbers: " + validNumbers); // ["1", "2", "4"]
System.out.println("Invalid Inputs: " + invalidInputs); // ["three", "five"]

In this example, Collectors.partitioningBy is used to partition the stream into two groups based on whether Integer.parseInt throws an exception or not.

Exceptions in Parallel Streams

When working with parallel streams, exceptions behave a bit differently. Since the stream operations can be executed concurrently on different threads, exceptions can be thrown from different threads.

To handle exceptions properly in parallel streams:

• Use proper synchronization and locking to avoid race conditions
• Catch exceptions from intermediate and terminal operations locally within lambdas to handle them at the source
• Avoid stateful lambdas which can cause thread interference

For example:

Stream<String> parallelStream = Stream.of("a", "b", "c").parallel(); 

parallelStream.map(s -> {
  synchronized(lock) {
    // stateful logic 
  }
  return s;
})
.forEach(s -> {
  System.out.println(s);
});

Here the shared state is properly synchronized while executing the map operation to avoid race conditions.

Sometimes it can be better to avoid parallel streams. Especially, if the operations have complex shared state or dependencies that are hard to synchronize.

There is a trick point about parallel streams. If an exception occurs in one of the stream items it only affects that item. Processing the other items continues. So if you want to process all items no matter what, You can use parallel stream.

Here is an example for this trick:

Stream<String> parallelStream = Stream.of("a", "b", "c").parallel();

parallelStream.map(s -> {
        if (s.equals("c")) {
            throw new RuntimeException("S is c");
        }

        return s;
    })
    .forEach(s -> {
        System.out.println(s);
    });

Use Optional to Handle Exceptions

You can consider using Optional to handle the potential absence of a value. It is useful when a stream operation might result in an exception.

List<String> list = Arrays.asList("1", "2", "3", "four", "5");
list.stream()
    .map(s -> {
        try {
            return Optional.of(Integer.parseInt(s));
        } catch (NumberFormatException e) {
            return Optional.empty(); // Return an empty Optional if parsing fails
        }
    })
    .forEach(opt -> opt.ifPresentOrElse(System.out::println, () -> System.out.println("Invalid")));

Best Practices for Exception Handling

Here are some best practices to follow for proper exception handling when using streams:

• Handle checked exceptions from stream initialization or intermediate operations at appropriate place by wrapping code in try-catch block
• Catch exceptions from lambda expressions of intermediate operations like map and flatMap locally within the lambda instead of later in the pipeline
• Avoid throwing raw exceptions from lambdas — wrap them appropriately in unchecked exceptions
• Keep lambdas stateless and share no state across threads when using parallel streams
• Add synchronization and locking properly when dealing with shared state in parallel streams
• Handle exceptions from terminal operations locally within the lambda
• Avoid relying on exception handling in terminal operation lambdas only
• Perform additional validation and error handling outside the stream pipeline if required

By following these practices, exceptions can be handled cleanly at the appropriate points within stream pipelines making the code more robust.

Custom Exception Handling Strategies

Sometimes we may want to follow a custom strategy for handling exceptions from streams. For example, we may want to -

• Log all exceptions and continue processing
• Collect all exceptions and handle them later
• Retry failed operations

For these scenarios, we can utilize some common patterns:

Logging Exceptions

To log all exceptions and continue processing, we can use a utility method:

public static void logException(Exception e) {
  // log exception
  // can rethrow if needed
}

And then wrap intermediate operations:

stream.map(s -> {
  try {
    // logic
  } catch (Exception e) {
    logException(e);
  }
  return s;
})

This will log the exception and continue processing the stream.

Collecting Exceptions

To accumulate all exceptions, we can collect them into a list:

List<Exception> exceptions = new ArrayList<>();

stream.map(s -> {
  try { 
    // logic 
  } catch (Exception e) {
    exceptions.add(e);
  }
  return s;
})
// handle exceptions list

The list contains all the exceptions while still processing the whole stream.

Retrying Failed Operations

We can also retry operations that fail by looping over them:

for(int i = 0; i < MAX_RETRIES; i++) {
  try {
    stream.forEach(s -> {
      // process 
    });
    break;
  } catch (Exception e) {    
    // retry 
  }
}

This will retry the terminal operation up to MAX_RETRIES times handling exceptions.

So in this way, we can implement custom exception handling approaches on streams when needed.

Exception handling requires special care when using Java streams to avoid masking exceptions or handling them at inappropriate places. Intermediate exceptions should be handled locally within lambdas instead of later during terminal operations.

Checked exceptions need to be handled by wrapping stream initialization in try-catch blocks. Parallel streams need extra synchronization and care to avoid thread interference.

By following best practices and custom strategies, exceptions can be handled cleanly without affecting the fluent streaming operations. Proper exception handling makes streams based processing more robust and production-ready.

👏 Thank You for Reading!

👨‍💼 I appreciate your time and hope you found this story insightful. If you enjoyed it, don’t forget to show your appreciation by clapping 👏 for the hard work!

📰 Keep the Knowledge Flowing by Sharing the Article!

✍ Feel free to share your feedback or opinions about the story. Your input helps me improve and create more valuable content for you.

✌ Stay Connected! 🚀 For more engaging articles, make sure to follow me on social media:

• Twitter
• Instagram

🔍 Explore More! 📖 Dive into a treasure trove of knowledge at Codimis. There’s always more to learn, and we’re here to help you on your journey of discovery.