StatelessPartitioner implements Partitioner<T>, Serializable { @Override public Collection<Partition<T>> definePartitions(Collection<Partition<T>> partitions, PartitioningContext context) { final int newPartitionCount = DefaultPartition.getRequiredPartitionCount(context, this.partitionCount); logger.debug("define partitions, partitionCount current {} requested {}", partitions.size(), newPartitionCount); DefaultPartition<T> partition = (DefaultPartition<T>)partitions.iterator().next(); Collection<Partition<T>> newPartitions; if (partitions.iterator().next().getStats() == null) { newPartitions = Lists.newArrayList(); for (int partitionCounter = 0; partitionCounter < newPartitionCount; partitionCounter++) { newPartitions.add(new DefaultPartition<>(partition.getPartitionedInstance())); } List<InputPort<?>> inputPortList = context.getInputPorts(); if (inputPortList != null && !inputPortList.isEmpty()) { DefaultPartition.assignPartitionKeys(newPartitions, inputPortList.iterator().next()); } } else { if (context.getParallelPartitionCount() != 0) { newPartitions = repartitionParallel(partitions, context); } else if (partition.getPartitionKeys().isEmpty()) { newPartitions = repartitionInputOperator(partitions); } else { newPartitions = repartition(partitions); } } logger.debug("new partition size {}", newPartitions.size()); return newPartitions; } StatelessPartitioner();  StatelessPartitioner(String value);  StatelessPartitioner(int partitionCount); void setPartitionCount(int partitionCount); int getPartitionCount(); @Override Collection<Partition<T>> definePartitions(Collection<Partition<T>> partitions, PartitioningContext context); @Override void partitioned(Map<Integer, Partition<T>> partitions); static Collection<Partition<T>> repartition(Collection<Partition<T>> partitions); static Collection<Partition<T>> repartitionInputOperator(Collection<Partition<T>> partitions); static Collection<Partition<T>> repartitionParallel(Collection<Partition<T>> partitions,
      PartitioningContext context);  }
@Test public void partition1Test() { DummyOperator dummyOperator = new DummyOperator(5); StatelessPartitioner<DummyOperator> statelessPartitioner = new StatelessPartitioner<>(); Collection<Partition<DummyOperator>> partitions = Lists.newArrayList(); DefaultPartition<DummyOperator> defaultPartition = new DefaultPartition<>(dummyOperator); partitions.add(defaultPartition); Collection<Partition<DummyOperator>> newPartitions = statelessPartitioner.definePartitions(partitions, new PartitioningContextImpl(null, 0)); Assert.assertEquals("Incorrect number of partitions", 1, newPartitions.size()); for (Partition<DummyOperator> partition : newPartitions) { Assert.assertEquals("Incorrect cloned value", 5, partition.getPartitionedInstance().getValue()); } }
@Test public void partition5Test() { DummyOperator dummyOperator = new DummyOperator(5); StatelessPartitioner<DummyOperator> statelessPartitioner = new StatelessPartitioner<>(5); Collection<Partition<DummyOperator>> partitions = Lists.newArrayList(); DefaultPartition<DummyOperator> defaultPartition = new DefaultPartition<>(dummyOperator); partitions.add(defaultPartition); Collection<Partition<DummyOperator>> newPartitions = statelessPartitioner.definePartitions(partitions, new PartitioningContextImpl(null, 0)); Assert.assertEquals("Incorrect number of partitions", 5, newPartitions.size()); for (Partition<DummyOperator> partition : newPartitions) { Assert.assertEquals("Incorrect cloned value", 5, partition.getPartitionedInstance().getValue()); } }
@Test public void testParallelPartitionScaleUP() { DummyOperator dummyOperator = new DummyOperator(5); StatelessPartitioner<DummyOperator> statelessPartitioner = new StatelessPartitioner<>(); Collection<Partition<DummyOperator>> partitions = Lists.newArrayList(); partitions.add(new DefaultPartition<>(dummyOperator)); Collection<Partition<DummyOperator>> newPartitions = statelessPartitioner.definePartitions(partitions, new PartitioningContextImpl(null, 5)); Assert.assertEquals("after partition", 5, newPartitions.size()); }
@Test public void testParallelPartitionScaleDown() { DummyOperator dummyOperator = new DummyOperator(5); StatelessPartitioner<DummyOperator> statelessPartitioner = new StatelessPartitioner<>(); Collection<Partition<DummyOperator>> partitions = Lists.newArrayList(); for (int i = 5; i-- > 0; ) { partitions.add(new DefaultPartition<>(dummyOperator)); } Collection<Partition<DummyOperator>> newPartitions = statelessPartitioner.definePartitions(partitions, new PartitioningContextImpl(null, 1)); Assert.assertEquals("after partition", 1, newPartitions.size()); }