how to use java streams efficiently and unit test it

Current situation:

I have already implemented the functionality using streams(code below), but considering that I'm not experienced with streams, I'm asking help to enhance my approach.

The problem:

A list of objects from class E should be processed using streams. These objects are built by me based on strings that are received as input. Each string contains key value pairs separated by a semicolon. I'm initially processing these strings and storing the data into a list of objects of class E, which will be streamed.

The E class has fields id, item1, item2, start time, end time, and value. For each object of class E with and id and a start time, there will always be another object with same id containing the end time

The object from E never stores start and end time on the same object.

While I stream the list of objects E, I need to calculate the average of the field value for item1 and another average for item2.

I must store the average for each item1 element in a Map<item1, average>.

I must store the average for each item2 element in a Map<item2, average>.

I need to store how many times each item1 took more than 300 seconds to be processed and store it into a Map<item1, howManyTimes>.

(not implemented yet) I need to store the item1 elements with value entries summing up an amount greater than 5000 in the last 5 hours and store it into a Set<item1>.

My questions: I currently stream a list of objects and then inside the forEach I created a block of code which is processing the logic requested above.

1. I would like to know if it's possible(and how) to process the data using other stream capabilities(filter/map,etc). Should I go on this way or keeping everything inside the forEach is ok?

2. I would like to have suggestions/examples of how to better design it to enable unit testing. Maybe create a class with methods containing the logic of the data processing and then unit test those methods?

public class Main {
    public static class LongRun {
        String item1;
        LocalDateTime startTime;

        public LongRun(String item1, LocalDateTime startTime) {
            this.item1 = item1;
            this.startTime = startTime;
        }

        public String getItem1() {
            return item1;
        }

        public LocalDateTime getStartTime() {
            return startTime;
        }
    }
    
    
    public static void main(String[] args) {
        List<E> listObject = new ArrayList<>();

        Map<String,Double> averageItem1 = new HashMap<>();
        Map<String,Integer> averageItem1Counter = new HashMap<>();

        Map<String,Double> averageItem2 = new HashMap<>();
        Map<String,Integer> averageItem2Counter = new HashMap<>();

        Map<Integer,LongRun> longRunId = new HashMap<>();
        Map<String,Integer> longRunIdCounter = new HashMap<>();

        Map<String,Double> item1AmountLastHour = new HashMap<>();

        int counter = 0;
        while(counter < 100) {
            E e = new E(stringWithKeyValuePairs);
            listObject.add(e);
            counter++;
        }

        listObject.stream().forEach(e -> {
            //average item1
            if(averageItem1.containsKey(e.getItem1())) {
                averageItem1Counter.put(e.getItem1(),averageItem1Counter.get(e.getItem1())+1);
                averageItem1.put(e.getItem1(),(averageItem1.get(e.getItem1())+e.getAmount())/averageItem1Counter.get(e.getItem1()));
            }
            else {
                if(e.getItem1() != null) {
                    averageItem1.put(e.getItem1(), e.getAmount());
                    averageItem1Counter.put(e.getItem1(),1);
                }
            }

            //average item2
            if(averageItem2.containsKey(e.getItem2())) {
                averageItem2.put(e.getItem2(),averageItem2.get(e.getItem2())+1);
                averageItem2.put(e.getItem2(),(averageItem2.get(e.getItem2())+e.getAmount())/averageItem2Counter.get(e.getItem2()));
            }
            else {
                if(e.getItem2() != null) {
                    averageItem2.put(e.getItem2(), e.getAmount());
                    averageItem2Counter.put(e.getItem2(),1);
                }
            }

            //long run 300 seconds
            if(longRunId.containsKey(e.getId())) {
                if(e.getEndTime() != null) {
                   Duration duration = Duration.between(longRunId.get(e.getId()).getStartTime(),e.getEndTime());

                   if(duration.getSeconds() > 300) {
                       if(longRunIdCounter.containsKey(e.getItem1())) {
                           longRunIdCounter.put(e.getItem1(),longRunIdCounter.get(e.getItem1())+1);
                       }
                       else {
                           if(e.getItem1() != null) {
                               longRunIdCounter.put(longRunId.get(e.getId()).getItem1(),1);
                           }
                       }
                   }
                }
            }
            else {
                if(e.getId() != null) {
                    longRunId.put(e.getId(), new LongRun(e.getItem1(),e.getStartTime()));
                }
            }
       });
    }
}