3

I have made a custom collector that uses a MessageDigest to create a hash. In general MessageDigest does not work in parallel. The issue I'm seeing is in the combiner() method. It is not possible to combine two MessageDigest objects. When I return null it seems to work but if I throw an UnsupportedOperationException it fails. What is the typical way to implement a collector that doesn't support parallel operations?

class ChecksumCollector implements Collector<String, MessageDigest, ByteBuffer> {
    private String algorithm;

    ChecksumCollector(final String algorithm) {
        this.algorithm = algorithm;
    }

    @Override
    public Supplier<MessageDigest> supplier() {
        return () -> {
            try {
                return MessageDigest.getInstance(algorithm);
            } catch (NoSuchAlgorithmException e) {
                throw new UnsupportedOperationException("Could not find MessageDigest for algorithm " + algorithm, e);
            }
        };
    }

    @Override
    public BiConsumer<MessageDigest, String> accumulator() {
        return (md, s) -> md.update(s.getBytes(StandardCharsets.UTF_8));
    }

    @Override
    public BinaryOperator<MessageDigest> combiner() {
        return null; //seems to work but hash may not be correct?
        //throw new UnsupportedOperationException(LineDuplicationHash.class.getSimpleName() + " does not support parallel streams");
    }

    @Override
    public Function<MessageDigest, ByteBuffer> finisher() {
        return md -> ByteBuffer.wrap(md.digest());
    }

    @Override
    public Set<Characteristics> characteristics() {
        return Set.of();
    }
}
0

3 Answers 3

4

A Collector’s BinaryOperator returned by combiner() will only be used when used for parallel streams, however the combiner() method itself will be invoked when calling Stream.collect() to retrieve that combiner, in the JDK’s implementation (see ReduceOps.makeRef(Collector)).

You thus have 2 options:

  • either return null, which would cause a NullPointerException if your collector is used in a parallel Stream, at the time the combiner needs to be used;
  • or return a BinaryOperator that actually throws the exception when called:
return (a, b) -> throw new UnsupportedOperationException(
    LineDuplicationHash.class.getSimpleName() + " does not support parallel streams");

This second option would be more friendly to the unknowing developer who later changes your pipeline to make it parallel.

2
  • 1
    Thanks! I wasn't sure how it would work and I wanted to be helpful by throwing the exception. I went with option 2. Feb 16, 2022 at 17:22
  • 1
    Note that this is just describing the typical use case. The specification makes no guaranty that a combiner is never used in a sequential context. The actual answer should be, if you can’t support all operations of a collector, don’t implement a collector.
    – Holger
    Feb 17, 2022 at 10:37
2

If you want to make this work you need to regain associativity by reducing indirectly over functions instead of directly over the message digest. Try this:

Collector<String, ?, ByteBuffer> checksumCollector = collectingAndThen(
    reducing(
        Function.identity(),
        s -> md -> {
            md.update(s.getBytes(StandardCharsets.UTF_8));
            return md;
        },
        (BinaryOperator<Function<MessageDigest, MessageDigest>>) Function::andThen),
    endo -> ByteBuffer.wrap(endo.apply(getMessageDigest("SHA-256")).digest()));

Where getMessageDigest() is a small helper method taking care of the checked exception:

static MessageDigest getMessageDigest(String algorithm) {
    try {
        return MessageDigest.getInstance(algorithm);
    } catch (NoSuchAlgorithmException e) {
        throw new UnsupportedOperationException("Could not find MessageDigest for algorithm " + algorithm, e);
    }    
}

Effectively this defers the actual calculation of the message digest to the finisher.

13
  • 1
    Please correct me if I'm wrong, but there's no method getMessageDigest() in the MessageDigest class. Probably it has to be static method getInstance() that throws NoSuchAlgorithmException, which is a checked exception and must be handled. Feb 16, 2022 at 22:30
  • 2
    Now it fits together. It's a very challenging code, a collector with a nested lambda that returns a function. But the fact that it does its job without entailing any auxiliary data structure is creditable. Definitely, that is an interesting approach. Feb 16, 2022 at 23:48
  • 2
    @JohnMercier I think your assumption is not correct. Parallel streams are utilizing ForkJoin framework under the hood and terminal operations like reduce(), findFirst(), collect(), forEachOrdered() respect the order in which threads are being assigned with their tasks. I.e. results from the different threads will be joint exactly in the same order these threads have obtained their part of the job. Feb 17, 2022 at 1:19
  • 2
    There's one thing about this code. It is obvious that all the data from the stream when collect gets executed has to be dumped into memory, like in the case of sorting. And I'm wondering how the memory consumption, in this case, will differ from the situation when stream elements are being collected into the array-based data structure (which has an upper hand of data locality)? A function that points to a string and in turn to another function which also has a reference to a string and points to another function - that resembles a linked list. Feb 17, 2022 at 4:07
  • 2
    "that resembles a linked list." @AlexanderIvanchenko, yes this is a drawback of this solution. TBH, in practice I would not use a collector for calculating a message digest because of the various complexities involved. I'd probably just us Stream.forEach() instead.
    – michid
    Feb 17, 2022 at 10:06
2

Although it doesn't make sense to work with MessageDigest in parallel, but if there would be some operation's in the pipeline that might be parallelized (and the data set has the potential to be large enough to pay to the overheads of using the parallel stream) you may consider the option of creating two versions of this collector.

The remedy for your sequential implementation is already provided in the Didier's answer. Providing a proper binary operator that throws an exception is far better than simply returning null.

For the parallel version of this collector data from the stream could be initially collected to an auxiliary mutable container and then the finisher function will populate the MessageDigest with data from the container. I want to reiterate that will make sense only if collect is presided by some operations that could be parallelized and input is significant enough otherwise it'll just cause unnecessary overheads.

That's how a parallel collector might look like

     Collector<String, ?, ByteBuffer> checksum =
                Collectors.collectingAndThen(Collectors.toList(), 
                                             list -> digestAndWrap(list, "SHA-512"));
    private static ByteBuffer digestAndWrap(List<String> list, String algorithm) {
        MessageDigest md;
        try {
            md = MessageDigest.getInstance(algorithm);
        } catch (NoSuchAlgorithmException e) {
            throw new UnsupportedOperationException("Could not find MessageDigest for algorithm " + algorithm, e);
        }
        for (String next: list) {
            md.update(next.getBytes(StandardCharsets.UTF_8));
        }
        return ByteBuffer.wrap(md.digest());
    }
8
  • @DidierL Thanks for pointing out that mistake, I've revised my solution. Feb 17, 2022 at 4:40
  • You’re welcome. I’m wondering why you are using toCollection(ArrayList::new) instead of just toList() though?
    – Didier L
    Feb 17, 2022 at 9:02
  • 2
    But your solution doesn’t rely on any specific of ArrayList, so if a future (or alternative) implementation uses a different (better) List implementation, why not allow it to use that?
    – Holger
    Feb 17, 2022 at 10:42
  • 2
    But ArrayList is not required for your task. The list in question is used for two tasks, first collecting the stream elements into it. There is no reason to assume that the authors of toList() would choose a list inappropriate for this task, as it is the very purpose of the collector. A different point of view is that by insisting on ArrayList, you claim to know better than the JDK developers, what’s best for this task. The second use is the for-each loop in your finisher. Every collection would do, but the JDK developers already know about “locality of data” and performance in general
    – Holger
    Feb 17, 2022 at 11:47
  • 2
    @Holger That makes sense, I agree that if at some point in time implementation of toList() will get changed the list that it'll provide will be a general-purpose implementation of the interface that in some cases will perform better than ArrayList. Thanks for your patience in explaining it. I've changed the code. Feb 17, 2022 at 12:57

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.