code: def test_window_all_aggregate(self):
      self.env.set_parallelism(1)
      data_stream = self.env.from_collection(
          [
              ("a", 1),
              ("a", 2),
              ("b", 3),
              ("a", 6),
              ("b", 8),
              ("b", 9),
              ("a", 15),
          ],
          type_info=Types.TUPLE([Types.STRING(), Types.INT()]),
      )  # type: DataStream
      watermark_strategy = WatermarkStrategy.for_monotonous_timestamps().with_timestamp_assigner(
          SecondColumnTimestampAssigner()
      )
  
      class MyAggregateFunction(AggregateFunction):
          def create_accumulator(self) -> Tuple[str, Dict[int, int]]:
              return "", {0: 0, 1: 0}
  
          def add(
              self,
              value: Tuple[str, int],
              accumulator: Tuple[str, Dict[int, int]],
          ) -> Tuple[str, Dict[int, int]]:
              number_map = accumulator[1]
              number_map[value[1] % 2] += 1
              return value[0], number_map
  
          def get_result(
              self, accumulator: Tuple[str, Dict[int, int]]
          ) -> Tuple[str, int]:
              number_map = accumulator[1]
              return accumulator[0], number_map[0] - number_map[1]
  
          def merge(
              self,
              acc_a: Tuple[str, Dict[int, int]],
              acc_b: Tuple[str, Dict[int, int]],
          ) -> Tuple[str, Dict[int, int]]:
              number_map_a = acc_a[1]
              number_map_b = acc_b[1]
              new_number_map = {
                  0: number_map_a[0] + number_map_b[0],
                  1: number_map_a[1] + number_map_b[1],
              }
              return acc_a[0], new_number_map
  
      data_stream.assign_timestamps_and_watermarks(
          watermark_strategy
      ).window_all(
          EventTimeSessionWindows.with_gap(Time.milliseconds(2))
      ).aggregate(
          MyAggregateFunction(),
          output_type=Types.TUPLE([Types.STRING(), Types.INT()]),
      ).add_sink(
          self.test_sink
      )
  
      self.env.execute("test_window_all_aggregate")
      results = self.test_sink.get_results()
      expected = ["(a,-1)", "(b,-1)", "(b,1)"]
      self.assert_equals_sorted(expected, results)
  
imports: from typing import Iterable, Tuple, Dict
  from pyflink.common import Configuration
  from pyflink.common.time import Time
  from pyflink.common.typeinfo import Types
  from pyflink.common.watermark_strategy import WatermarkStrategy, TimestampAssigner
  from pyflink.datastream.data_stream import DataStream
  from pyflink.datastream.functions import ProcessWindowFunction, WindowFunction, AggregateFunction, ProcessAllWindowFunction
  from pyflink.datastream.output_tag import OutputTag
  from pyflink.datastream.window import TumblingEventTimeWindows, SlidingEventTimeWindows, EventTimeSessionWindows, CountSlidingWindowAssigner, SessionWindowTimeGapExtractor, CountWindow, PurgingTrigger, EventTimeTrigger, TimeWindow, GlobalWindows, CountTrigger
  from pyflink.datastream.tests.test_util import DataStreamTestSinkFunction
  from pyflink.java_gateway import get_gateway
  from pyflink.testing.test_case_utils import PyFlinkStreamingTestCase
  from pyflink.util.java_utils import get_j_env_configuration
label_desc: Use image node_id 21 for calling the WindowTests obj's underlying member method code with example usage: obj.test_window_all_aggregate() and returns: str,  dict, value,  number_map, accumulator,  unknown, acc_a,  new_number_map