test_class.declarator: topicName = "topic"
test_class.modifier: private
test_class.original_string: private String topicName = "topic";
test_class.type: String
test_class.var_name: topicName
test_class.declarator: intSerializer = new IntegerSerializer()
test_class.modifier: private
test_class.original_string: private IntegerSerializer intSerializer = new IntegerSerializer();
test_class.type: IntegerSerializer
test_class.var_name: intSerializer
test_class.declarator: stringSerializer = new StringSerializer()
test_class.modifier: private
test_class.original_string: private StringSerializer stringSerializer = new StringSerializer();
test_class.type: StringSerializer
test_class.var_name: stringSerializer
test_class.declarator: infos = Arrays.asList(
              new PartitionInfo(topicName, 0, Node.noNode(), new Node[0], new Node[0]),
              new PartitionInfo(topicName, 1, Node.noNode(), new Node[0], new Node[0]),
              new PartitionInfo(topicName, 2, Node.noNode(), new Node[0], new Node[0]),
              new PartitionInfo(topicName, 3, Node.noNode(), new Node[0], new Node[0]),
              new PartitionInfo(topicName, 4, Node.noNode(), new Node[0], new Node[0]),
              new PartitionInfo(topicName, 5, Node.noNode(), new Node[0], new Node[0])
      )
test_class.modifier: private
test_class.original_string: private List<PartitionInfo> infos = Arrays.asList(
              new PartitionInfo(topicName, 0, Node.noNode(), new Node[0], new Node[0]),
              new PartitionInfo(topicName, 1, Node.noNode(), new Node[0], new Node[0]),
              new PartitionInfo(topicName, 2, Node.noNode(), new Node[0], new Node[0]),
              new PartitionInfo(topicName, 3, Node.noNode(), new Node[0], new Node[0]),
              new PartitionInfo(topicName, 4, Node.noNode(), new Node[0], new Node[0]),
              new PartitionInfo(topicName, 5, Node.noNode(), new Node[0], new Node[0])
      );
test_class.type: List<PartitionInfo>
test_class.var_name: infos
test_class.declarator: cluster = new Cluster(Collections.singletonList(Node.noNode()), infos, Collections.<String>emptySet())
test_class.modifier: private
test_class.original_string: private Cluster cluster = new Cluster(Collections.singletonList(Node.noNode()), infos, Collections.<String>emptySet());
test_class.type: Cluster
test_class.var_name: cluster
test_class.file: streams/src/test/java/org/apache/kafka/streams/kstream/internals/WindowedStreamPartitionerTest.java
test_class.identifier: WindowedStreamPartitionerTest
test_case.body: @Test
      public void testCopartitioning() {
  
          Random rand = new Random();
  
          DefaultPartitioner defaultPartitioner = new DefaultPartitioner();
  
          WindowedSerializer<Integer> windowedSerializer = new WindowedSerializer<>(intSerializer);
          WindowedStreamPartitioner<Integer, String> streamPartitioner = new WindowedStreamPartitioner<>(windowedSerializer);
  
          for (int k = 0; k < 10; k++) {
              Integer key = rand.nextInt();
              byte[] keyBytes = intSerializer.serialize(topicName, key);
  
              String value = key.toString();
              byte[] valueBytes = stringSerializer.serialize(topicName, value);
  
              Integer expected = defaultPartitioner.partition("topic", key, keyBytes, value, valueBytes, cluster);
  
              for (int w = 0; w < 10; w++) {
                  TimeWindow window = new TimeWindow(10 * w, 20 * w);
  
                  Windowed<Integer> windowedKey = new Windowed<>(key, window);
                  Integer actual = streamPartitioner.partition(windowedKey, value, infos.size());
  
                  assertEquals(expected, actual);
              }
          }
      }
test_case.class_method_signature: WindowedStreamPartitionerTest.testCopartitioning()
test_case.full_signature: @Test public void testCopartitioning()
test_case.identifier: testCopartitioning
test_case.invocations: nextInt serialize toString serialize partition partition size assertEquals
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testCopartitioning()
focal_class.declarator: serializer
focal_class.modifier: private final
focal_class.original_string: private final WindowedSerializer<K> serializer;
focal_class.type: WindowedSerializer<K>
focal_class.var_name: serializer
focal_class.file: streams/src/main/java/org/apache/kafka/streams/kstream/internals/WindowedStreamPartitioner.java
focal_class.identifier: WindowedStreamPartitioner
focal_class.interfaces: implements StreamPartitioner<Windowed<K>, V>
focal_class.class_method_signature: WindowedStreamPartitioner.WindowedStreamPartitioner(WindowedSerializer<K> serializer)
focal_class.full_signature: public  WindowedStreamPartitioner(WindowedSerializer<K> serializer)
focal_class.identifier: WindowedStreamPartitioner
focal_class.modifiers: public
focal_class.parameters: (WindowedSerializer<K> serializer)
focal_class.signature:  WindowedStreamPartitioner(WindowedSerializer<K> serializer)
focal_class.class_method_signature: WindowedStreamPartitioner.partition(Windowed<K> windowedKey, V value, int numPartitions)
focal_class.full_signature: public Integer partition(Windowed<K> windowedKey, V value, int numPartitions)
focal_class.identifier: partition
focal_class.modifiers: public
focal_class.parameters: (Windowed<K> windowedKey, V value, int numPartitions)
focal_class.return: Integer
focal_class.signature: Integer partition(Windowed<K> windowedKey, V value, int numPartitions)
focal_class.class_method_signature: WindowedStreamPartitioner.toPositive(int number)
focal_class.full_signature: private static int toPositive(int number)
focal_class.identifier: toPositive
focal_class.modifiers: private static
focal_class.parameters: (int number)
focal_class.return: int
focal_class.signature: int toPositive(int number)
focal_method.body: public Integer partition(Windowed<K> windowedKey, V value, int numPartitions) {
          byte[] keyBytes = serializer.serializeBaseKey(null, windowedKey);
  
          // hash the keyBytes to choose a partition
          return toPositive(Utils.murmur2(keyBytes)) % numPartitions;
      }
focal_method.class_method_signature: WindowedStreamPartitioner.partition(Windowed<K> windowedKey, V value, int numPartitions)
focal_method.full_signature: public Integer partition(Windowed<K> windowedKey, V value, int numPartitions)
focal_method.identifier: partition
focal_method.invocations: serializeBaseKey toPositive murmur2
focal_method.modifiers: public
focal_method.parameters: (Windowed<K> windowedKey, V value, int numPartitions)
focal_method.return: Integer
focal_method.signature: Integer partition(Windowed<K> windowedKey, V value, int numPartitions)