import org.apache.samza.util.SinglePartitionWithoutOffsetsSystemAdmin; //导入依赖的package包/类
@Test
public void testCoordinatorStreamSystemConsumer() {
  Map<String, String> expectedConfig = new LinkedHashMap<String, String>();
  expectedConfig.put("job.id", "1234");
  SystemStream systemStream = new SystemStream("system", "stream");
  MockSystemConsumer systemConsumer = new MockSystemConsumer(new SystemStreamPartition(systemStream, new Partition(0)));
  CoordinatorStreamSystemConsumer consumer = new CoordinatorStreamSystemConsumer(systemStream, systemConsumer, new SinglePartitionWithoutOffsetsSystemAdmin());
  assertEquals(0, systemConsumer.getRegisterCount());
  consumer.register();
  assertEquals(1, systemConsumer.getRegisterCount());
  assertFalse(systemConsumer.isStarted());
  consumer.start();
  assertTrue(systemConsumer.isStarted());
  try {
    consumer.getConfig();
    fail("Should have failed when retrieving config before bootstrapping.");
  } catch (SamzaException e) {
    // Expected.
  }
  consumer.bootstrap();
  assertEquals(expectedConfig, consumer.getConfig());
  assertFalse(systemConsumer.isStopped());
  consumer.stop();
  assertTrue(systemConsumer.isStopped());
}
 

import org.apache.samza.util.SinglePartitionWithoutOffsetsSystemAdmin; //导入依赖的package包/类
@Test
public void testCoordinatorStreamSystemConsumerRegisterOnceOnly() throws Exception {
  Map<String, String> expectedConfig = new LinkedHashMap<String, String>();
  expectedConfig.put("job.id", "1234");
  SystemStream systemStream = new SystemStream("system", "stream");
  MockSystemConsumer systemConsumer = new MockSystemConsumer(new SystemStreamPartition(systemStream, new Partition(0)));
  CoordinatorStreamSystemConsumer consumer = new CoordinatorStreamSystemConsumer(systemStream, systemConsumer, new SinglePartitionWithoutOffsetsSystemAdmin());
  assertEquals(0, systemConsumer.getRegisterCount());
  consumer.register();
  assertEquals(1, systemConsumer.getRegisterCount());
  assertFalse(systemConsumer.isStarted());
  consumer.start();
  assertTrue(systemConsumer.isStarted());
  consumer.register();
  assertEquals(1, systemConsumer.getRegisterCount());
}
 

import org.apache.samza.util.SinglePartitionWithoutOffsetsSystemAdmin; //导入依赖的package包/类
/**
 * Verify that if a particular key-value is written, then another, then the original again,
 * that the original occurs last in the set.
 */
@Test
public void testOrderKeyRewrite() throws InterruptedException {
  final SystemStream systemStream = new SystemStream("system", "stream");
  final SystemStreamPartition ssp = new SystemStreamPartition(systemStream, new Partition(0));
  final SystemConsumer systemConsumer = mock(SystemConsumer.class);

  final List<IncomingMessageEnvelope> list = new ArrayList<>();
  SetConfig setConfig1 = new SetConfig("source", "key1", "value1");
  SetConfig setConfig2 = new SetConfig("source", "key1", "value2");
  SetConfig setConfig3 = new SetConfig("source", "key1", "value1");
  list.add(createIncomingMessageEnvelope(setConfig1, ssp));
  list.add(createIncomingMessageEnvelope(setConfig2, ssp));
  list.add(createIncomingMessageEnvelope(setConfig3, ssp));
  Map<SystemStreamPartition, List<IncomingMessageEnvelope>> messages = new HashMap<SystemStreamPartition, List<IncomingMessageEnvelope>>() {
    {
      put(ssp, list);
    }
  };
  when(systemConsumer.poll(anySet(), anyLong())).thenReturn(messages, Collections.<SystemStreamPartition, List<IncomingMessageEnvelope>>emptyMap());

  CoordinatorStreamSystemConsumer consumer = new CoordinatorStreamSystemConsumer(systemStream, systemConsumer, new SinglePartitionWithoutOffsetsSystemAdmin());

  consumer.bootstrap();

  Set<CoordinatorStreamMessage> bootstrappedMessages = consumer.getBoostrappedStream();

  assertEquals(2, bootstrappedMessages.size()); // First message should have been removed as a duplicate
  CoordinatorStreamMessage[] coordinatorStreamMessages = bootstrappedMessages.toArray(new CoordinatorStreamMessage[2]);
  assertEquals(setConfig2, coordinatorStreamMessages[0]);
  assertEquals(setConfig3, coordinatorStreamMessages[1]); //Config 3 MUST be the last message, not config 2
}
 

import org.apache.samza.util.SinglePartitionWithoutOffsetsSystemAdmin; //导入依赖的package包/类
@Override
public SystemAdmin getAdmin(String systemName, Config config) {
  return new SinglePartitionWithoutOffsetsSystemAdmin();
}
 

import org.apache.samza.util.SinglePartitionWithoutOffsetsSystemAdmin; //导入依赖的package包/类
public SystemAdmin getAdmin(String systemName, Config config) {
  return new SinglePartitionWithoutOffsetsSystemAdmin();
}
 

import org.apache.samza.util.SinglePartitionWithoutOffsetsSystemAdmin; //导入依赖的package包/类
@Override
public SystemAdmin getAdmin(String systemName, Config config) {
	return new SinglePartitionWithoutOffsetsSystemAdmin();
}