### Question:
ComposeOffsetStrategy implements OffsetStrategy { @Override public Map<String, Object> toPartitionMap(final SourceMetadata metadata) { Map<String, Object> offset = new LinkedHashMap<>(); for (GenericOffsetStrategy strategy : strategies) { strategy.addAttributeToPartitionMap(metadata, offset); } return offset; } ComposeOffsetStrategy(final String offsetStrategyString); @Override Map<String, Object> toPartitionMap(final SourceMetadata metadata); @Override String toString();  }### Answer:
@Test(expected = IllegalArgumentException.class) public void should_throw_illegal_argument_given_empty_strategy() { new ComposeOffsetStrategy("").toPartitionMap(metadata); }
@Test(expected = IllegalArgumentException.class) public void should_throw_illegal_argument_given_unknown_strategy() { new ComposeOffsetStrategy("dummy").toPartitionMap(metadata); }
@Test(expected = NullPointerException.class) public void should_throw_npe_given_unknown_strategy() { new ComposeOffsetStrategy(null).toPartitionMap(metadata); }
@Test public void should_get_offset_based_on_path() { Map<String, Object> result = new ComposeOffsetStrategy("PATH").toPartitionMap(metadata); Assert.assertEquals(1, result.size()); Assert.assertEquals("/tmp/path", result.get("path")); }
@Test public void should_get_offset_based_on_hash() { Map<String, Object> result = new ComposeOffsetStrategy("HASH").toPartitionMap(metadata); Assert.assertEquals(1, result.size()); Assert.assertEquals(789L, result.get("hash")); }
@Test public void should_get_offset_based_on_modified() { Map<String, Object> result = new ComposeOffsetStrategy("LASTMODIFIED").toPartitionMap(metadata); Assert.assertEquals(1, result.size()); Assert.assertEquals(123L, result.get("lastmodified")); }
@Test public void should_get_offset_based_on_name() { Map<String, Object> result = new ComposeOffsetStrategy("NAME").toPartitionMap(metadata); Assert.assertEquals(1, result.size()); Assert.assertEquals("test", result.get("name")); }
@Test public void should_get_composed_offset_based_on_path_and_hash() { Map<String, Object> result = new ComposeOffsetStrategy("PATH+HASH").toPartitionMap(metadata); Assert.assertEquals(2, result.size()); Assert.assertEquals("/tmp/path", result.get("path")); Assert.assertEquals(789L, result.get("hash")); }
@Test public void should_support_strategies_in_any_order() { final String o1 = new ComposeOffsetStrategy("PATH+HASH").toPartitionMap(metadata) .entrySet() .stream() .map(e -> e.getKey() + "=" + e.getValue()) .collect(Collectors.joining(",")); final String o2 = new ComposeOffsetStrategy("HASH+PATH").toPartitionMap(metadata) .entrySet() .stream() .map(e -> e.getKey() + "=" + e.getValue()) .collect(Collectors.joining(",")); Assert.assertEquals(o1, o2); }