ShardManager { public Map<String, String> loadInitialShardPositions() { return shardPos; }  Map<String, String> loadInitialShardPositions(); void updatePositions(Map<String, String> shardPositions);  }
@Test public void testShardManager() throws Exception { latch = new CountDownLatch(3); KinesisTestProducer p = new KinesisTestProducer(streamName, true); p.setSendCount(totalCount); p.run(); LocalMode lma = LocalMode.newInstance(); DAG dag = lma.getDAG(); KinesisUtil.getInstance().setClient(client); KinesisStringInputOperator node = dag.addOperator("Kinesis consumer", KinesisStringInputOperator.class); node.setAccessKey(credentials.getCredentials().getAWSSecretKey()); node.setSecretKey(credentials.getCredentials().getAWSAccessKeyId()); node.setStreamName(streamName); ShardManager tfm = new ShardManager(); node.setShardManager(tfm); node.setStrategy(AbstractKinesisInputOperator.PartitionStrategy.MANY_TO_ONE.toString()); node.setRepartitionInterval(-1); KinesisConsumer consumer = new KinesisConsumer(); consumer.setStreamName(streamName); consumer.setInitialOffset("earliest"); node.setConsumer(consumer); CollectorModule collector = dag.addOperator("RecordCollector", new CollectorModule()); dag.addStream("Kinesis Records", node.outputPort, collector.inputPort).setLocality(Locality.CONTAINER_LOCAL); final LocalMode.Controller lc = lma.getController(); lc.setHeartbeatMonitoringEnabled(true); lc.runAsync(); latch.await(10000, TimeUnit.MILLISECONDS); assertEquals("ShardPos Size", 2, node.getShardManager().loadInitialShardPositions().size()); Iterator ite = node.getShardManager().loadInitialShardPositions().entrySet().iterator(); Entry e = (Entry)ite.next(); assertNotEquals("Record Seq No in Shard Id 1", "", e.getValue()); e = (Entry)ite.next(); assertNotEquals("Record Seq No in Shard Id 2", "", e.getValue()); assertEquals("Tuple count", totalCount, collectedTuples.size()); logger.debug(String.format("Number of emitted tuples: %d -> %d", collectedTuples.size(), totalCount)); lc.shutdown(); }