InMemorySystemProducer implements SystemProducer { @Override public void send(String source, OutgoingMessageEnvelope envelope) { Object key = envelope.getKey(); Object message = envelope.getMessage(); Object partitionKey; if (envelope.getPartitionKey() != null) { partitionKey = envelope.getPartitionKey(); } else if (key != null) { partitionKey = key; } else { partitionKey = message; } Preconditions.checkNotNull(partitionKey, "Failed to compute partition key for the message: " + envelope); int partition = Math.abs(hashCode(partitionKey)) % memoryManager.getPartitionCountForSystemStream(envelope.getSystemStream()); SystemStreamPartition ssp = new SystemStreamPartition(envelope.getSystemStream(), new Partition(partition)); memoryManager.put(ssp, key, message); } InMemorySystemProducer(String systemName, InMemoryManager manager); @Override void start(); @Override void stop(); @Override void register(String source); @Override void send(String source, OutgoingMessageEnvelope envelope); void send(IncomingMessageEnvelope envelope); @Override void flush(String source);  }
@Test public void testPartition() { doReturn(1000).when(inMemoryManager).getPartitionCountForSystemStream(any()); doAnswer(new Answer<Void>() { int partitionOfFirstMessage = -1; int partitionOfSecondMessage = -2; @Override public Void answer(InvocationOnMock invocation) throws Throwable { SystemStreamPartition ssp = invocation.getArgumentAt(0, SystemStreamPartition.class); if (partitionOfFirstMessage == -1) { partitionOfFirstMessage = ssp.getPartition().getPartitionId(); } else { partitionOfSecondMessage = ssp.getPartition().getPartitionId(); Assert.assertEquals(partitionOfFirstMessage, partitionOfSecondMessage); testFinished = true; } return null; } }).when(inMemoryManager).put(any(), any(), any()); byte[] key1 = new byte[]{1, 2, 3}; byte[] key2 = new byte[]{1, 2, 3}; SystemStream systemStream = new SystemStream("TestSystem", "TestStream"); OutgoingMessageEnvelope outgoingMessageEnvelope1 = new OutgoingMessageEnvelope(systemStream, key1, null); OutgoingMessageEnvelope outgoingMessageEnvelope2 = new OutgoingMessageEnvelope(systemStream, key2, null); inMemorySystemProducer.send("TestSource", outgoingMessageEnvelope1); inMemorySystemProducer.send("TestSource", outgoingMessageEnvelope2); Assert.assertTrue(testFinished); }