public void convertOffset(String brokerList, String topic, Map<String, String> properties, long warehouseId)
            throws UserException {
        List<Integer> beginningPartitions = Lists.newArrayList();
        List<Integer> endPartitions = Lists.newArrayList();
        for (Map.Entry<Integer, Long> entry : partitionIdToOffset.entrySet()) {
            Integer part = entry.getKey();
            Long offset = entry.getValue();
            if (offset == -2L) {
                beginningPartitions.add(part);
            }
            if (offset == -1L) {
                endPartitions.add(part);
            }
        }

        if (beginningPartitions.size() > 0) {
            Map<Integer, Long> partOffsets = KafkaUtil
                    .getBeginningOffsets(brokerList, topic, ImmutableMap.copyOf(properties), beginningPartitions, warehouseId);
            partitionIdToOffset.putAll(partOffsets);
        }
        if (endPartitions.size() > 0) {
            Map<Integer, Long> partOffsets =
                    KafkaUtil.getLatestOffsets(brokerList, topic, ImmutableMap.copyOf(properties), endPartitions, warehouseId);
            partitionIdToOffset.putAll(partOffsets);
        }
    }
@Test
    public void testConvertOffset() throws Exception {
        new MockUp<KafkaUtil>() {
            @Mock
            public Map<Integer, Long> getLatestOffsets(String brokerList, String topic,
                                                       ImmutableMap<String, String> properties,
                                                       List<Integer> partitions,
                                                       long warehouseId) throws UserException {
                Map<Integer, Long> result = Maps.newHashMap();
                result.put(0, 100L);
                return result;
            }

            @Mock
            public Map<Integer, Long> getBeginningOffsets(String brokerList, String topic,
                                                          ImmutableMap<String, String> properties,
                                                          List<Integer> partitions,
                                                          long warehouseId) throws UserException {
                Map<Integer, Long> result = Maps.newHashMap();
                result.put(1, 1L);
                return result;
            }
        };

        KafkaProgress progress = new KafkaProgress();
        // modify offset while paused when partition is not ready
        try {
            List<Pair<Integer, Long>> partitionToOffset = new ArrayList<>();
            partitionToOffset.add(new Pair<>(3, 20L));
            progress.modifyOffset(partitionToOffset);
        } catch (DdlException e) {
            Assert.assertEquals("The specified partition 3 is not in the consumed partitions", e.getMessage());
        }

        progress.addPartitionOffset(new Pair<>(0, -1L));
        progress.addPartitionOffset(new Pair<>(1, -2L));
        progress.addPartitionOffset(new Pair<>(2, 10L));
        progress.addPartitionOffset(new Pair<>(3, 10L));
        progress.convertOffset("127.0.0.1:9020", "topic", Maps.newHashMap(), WarehouseManager.DEFAULT_WAREHOUSE_ID);

        List<Pair<Integer, Long>> partitionToOffset = new ArrayList<>();
        partitionToOffset.add(new Pair<>(3, 20L));
        progress.modifyOffset(partitionToOffset);
        Assert.assertEquals(4, partitionToOffset.size());

        Assert.assertEquals(100L, (long) progress.getOffsetByPartition(0));
        Assert.assertEquals(1L, (long) progress.getOffsetByPartition(1));
        Assert.assertEquals(10L, (long) progress.getOffsetByPartition(2));
        Assert.assertEquals(20L, (long) progress.getOffsetByPartition(3));
    }