### Question:
RedisDataStore implements DataStore { @Override public Set<Ack> removeAcknowledged(final String uaid, final Set<Ack> acks) { final Jedis jedis = jedisPool.getResource(); try { for (Ack ack : acks) { jedis.del(ackLookupKey(ack.getChannelId())); jedis.srem(acksLookupKey(uaid), ack.getChannelId()); } return getUnacknowledged(uaid); } finally { jedisPool.returnResource(jedis); } } RedisDataStore(final String host, final int port); @Override void savePrivateKeySalt(final byte[] salt); @Override byte[] getPrivateKeySalt(); @Override boolean saveChannel(final Channel channel); @Override void removeChannels(final Set<String> channelIds); @Override Channel getChannel(final String channelId); @Override Set<String> getChannelIds(final String uaid); @Override void removeChannels(final String uaid); @Override String updateVersion(final String endpointToken, final long newVersion); @Override String saveUnacknowledged(final String channelId, final long version); @Override Set<Ack> getUnacknowledged(final String uaid); @Override Set<Ack> removeAcknowledged(final String uaid, final Set<Ack> acks);  }### Answer:
@Test public void removeAcknowledged() { final RedisDataStore store = newRedisDataStore(); final String uaid = UUIDUtil.newUAID(); final Channel channel1 = newChannel2(uaid); final Channel channel2 = newChannel2(uaid); store.saveChannel(channel1); store.saveChannel(channel2); store.saveUnacknowledged(channel1.getChannelId(), channel1.getVersion()); store.saveUnacknowledged(channel2.getChannelId(), channel2.getVersion()); final Set<Ack> acks = asSet(new AckImpl(channel1.getChannelId(), channel1.getVersion())); final Set<Ack> unacknowledged = store.removeAcknowledged(channel1.getUAID(), acks); assertThat(unacknowledged.size(), is(1)); assertThat(unacknowledged, hasItem(new AckImpl(channel2.getChannelId(), channel2.getVersion()))); }