@Ignore @Test public void testExecute() throws SubCommandException { ResetOffsetByTimeCommand cmd = new ResetOffsetByTimeCommand(); Options options = ServerUtil.buildCommandlineOptions(new Options()); String[] subargs = new String[] {"-g default-group", "-t unit-test", "-s 1412131213231", "-f false"}; final CommandLine commandLine = ServerUtil.parseCmdLine("mqadmin " + cmd.commandName(), subargs, cmd.buildCommandlineOptions(options), new PosixParser()); cmd.execute(commandLine, options, null); }
@Override public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) throws SubCommandException { DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook); defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis())); try { String group = commandLine.getOptionValue("g").trim(); String topic = commandLine.getOptionValue("t").trim(); String timeStampStr = commandLine.getOptionValue("s").trim(); long timestamp = timeStampStr.equals("now") ? System.currentTimeMillis() : 0; try { if (timestamp == 0) { timestamp = Long.parseLong(timeStampStr); } } catch (NumberFormatException e) { timestamp = UtilAll.parseDate(timeStampStr, UtilAll.YYYY_MM_DD_HH_MM_SS_SSS).getTime(); } boolean force = true; if (commandLine.hasOption('f')) { force = Boolean.valueOf(commandLine.getOptionValue("f").trim()); } boolean isC = false; if (commandLine.hasOption('c')) { isC = true; } defaultMQAdminExt.start(); Map<MessageQueue, Long> offsetTable; try { offsetTable = defaultMQAdminExt.resetOffsetByTimestamp(topic, group, timestamp, force, isC); } catch (MQClientException e) { if (ResponseCode.CONSUMER_NOT_ONLINE == e.getResponseCode()) { ResetOffsetByTimeOldCommand.resetOffset(defaultMQAdminExt, group, topic, timestamp, force, timeStampStr); return; } throw e; } System.out.printf("rollback consumer offset by specified group[%s], topic[%s], force[%s], timestamp(string)[%s], timestamp(long)[%s]%n", group, topic, force, timeStampStr, timestamp); System.out.printf("%-40s %-40s %-40s%n", "#brokerName", "#queueId", "#offset"); Iterator<Map.Entry<MessageQueue, Long>> iterator = offsetTable.entrySet().iterator(); while (iterator.hasNext()) { Map.Entry<MessageQueue, Long> entry = iterator.next(); System.out.printf("%-40s %-40d %-40d%n", UtilAll.frontStringAtLeast(entry.getKey().getBrokerName(), 32), entry.getKey().getQueueId(), entry.getValue()); } } catch (Exception e) { throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e); } finally { defaultMQAdminExt.shutdown(); } }
ResetOffsetByTimeCommand implements SubCommand { @Override public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) throws SubCommandException { DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook); defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis())); try { String group = commandLine.getOptionValue("g").trim(); String topic = commandLine.getOptionValue("t").trim(); String timeStampStr = commandLine.getOptionValue("s").trim(); long timestamp = timeStampStr.equals("now") ? System.currentTimeMillis() : 0; try { if (timestamp == 0) { timestamp = Long.parseLong(timeStampStr); } } catch (NumberFormatException e) { timestamp = UtilAll.parseDate(timeStampStr, UtilAll.YYYY_MM_DD_HH_MM_SS_SSS).getTime(); } boolean force = true; if (commandLine.hasOption('f')) { force = Boolean.valueOf(commandLine.getOptionValue("f").trim()); } boolean isC = false; if (commandLine.hasOption('c')) { isC = true; } defaultMQAdminExt.start(); Map<MessageQueue, Long> offsetTable; try { offsetTable = defaultMQAdminExt.resetOffsetByTimestamp(topic, group, timestamp, force, isC); } catch (MQClientException e) { if (ResponseCode.CONSUMER_NOT_ONLINE == e.getResponseCode()) { ResetOffsetByTimeOldCommand.resetOffset(defaultMQAdminExt, group, topic, timestamp, force, timeStampStr); return; } throw e; } System.out.printf("rollback consumer offset by specified group[%s], topic[%s], force[%s], timestamp(string)[%s], timestamp(long)[%s]%n", group, topic, force, timeStampStr, timestamp); System.out.printf("%-40s %-40s %-40s%n", "#brokerName", "#queueId", "#offset"); Iterator<Map.Entry<MessageQueue, Long>> iterator = offsetTable.entrySet().iterator(); while (iterator.hasNext()) { Map.Entry<MessageQueue, Long> entry = iterator.next(); System.out.printf("%-40s %-40d %-40d%n", UtilAll.frontStringAtLeast(entry.getKey().getBrokerName(), 32), entry.getKey().getQueueId(), entry.getValue()); } } catch (Exception e) { throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e); } finally { defaultMQAdminExt.shutdown(); } } }
ResetOffsetByTimeCommand implements SubCommand { @Override public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) throws SubCommandException { DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook); defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis())); try { String group = commandLine.getOptionValue("g").trim(); String topic = commandLine.getOptionValue("t").trim(); String timeStampStr = commandLine.getOptionValue("s").trim(); long timestamp = timeStampStr.equals("now") ? System.currentTimeMillis() : 0; try { if (timestamp == 0) { timestamp = Long.parseLong(timeStampStr); } } catch (NumberFormatException e) { timestamp = UtilAll.parseDate(timeStampStr, UtilAll.YYYY_MM_DD_HH_MM_SS_SSS).getTime(); } boolean force = true; if (commandLine.hasOption('f')) { force = Boolean.valueOf(commandLine.getOptionValue("f").trim()); } boolean isC = false; if (commandLine.hasOption('c')) { isC = true; } defaultMQAdminExt.start(); Map<MessageQueue, Long> offsetTable; try { offsetTable = defaultMQAdminExt.resetOffsetByTimestamp(topic, group, timestamp, force, isC); } catch (MQClientException e) { if (ResponseCode.CONSUMER_NOT_ONLINE == e.getResponseCode()) { ResetOffsetByTimeOldCommand.resetOffset(defaultMQAdminExt, group, topic, timestamp, force, timeStampStr); return; } throw e; } System.out.printf("rollback consumer offset by specified group[%s], topic[%s], force[%s], timestamp(string)[%s], timestamp(long)[%s]%n", group, topic, force, timeStampStr, timestamp); System.out.printf("%-40s %-40s %-40s%n", "#brokerName", "#queueId", "#offset"); Iterator<Map.Entry<MessageQueue, Long>> iterator = offsetTable.entrySet().iterator(); while (iterator.hasNext()) { Map.Entry<MessageQueue, Long> entry = iterator.next(); System.out.printf("%-40s %-40d %-40d%n", UtilAll.frontStringAtLeast(entry.getKey().getBrokerName(), 32), entry.getKey().getQueueId(), entry.getValue()); } } catch (Exception e) { throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e); } finally { defaultMQAdminExt.shutdown(); } }  }
ResetOffsetByTimeCommand implements SubCommand { @Override public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) throws SubCommandException { DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook); defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis())); try { String group = commandLine.getOptionValue("g").trim(); String topic = commandLine.getOptionValue("t").trim(); String timeStampStr = commandLine.getOptionValue("s").trim(); long timestamp = timeStampStr.equals("now") ? System.currentTimeMillis() : 0; try { if (timestamp == 0) { timestamp = Long.parseLong(timeStampStr); } } catch (NumberFormatException e) { timestamp = UtilAll.parseDate(timeStampStr, UtilAll.YYYY_MM_DD_HH_MM_SS_SSS).getTime(); } boolean force = true; if (commandLine.hasOption('f')) { force = Boolean.valueOf(commandLine.getOptionValue("f").trim()); } boolean isC = false; if (commandLine.hasOption('c')) { isC = true; } defaultMQAdminExt.start(); Map<MessageQueue, Long> offsetTable; try { offsetTable = defaultMQAdminExt.resetOffsetByTimestamp(topic, group, timestamp, force, isC); } catch (MQClientException e) { if (ResponseCode.CONSUMER_NOT_ONLINE == e.getResponseCode()) { ResetOffsetByTimeOldCommand.resetOffset(defaultMQAdminExt, group, topic, timestamp, force, timeStampStr); return; } throw e; } System.out.printf("rollback consumer offset by specified group[%s], topic[%s], force[%s], timestamp(string)[%s], timestamp(long)[%s]%n", group, topic, force, timeStampStr, timestamp); System.out.printf("%-40s %-40s %-40s%n", "#brokerName", "#queueId", "#offset"); Iterator<Map.Entry<MessageQueue, Long>> iterator = offsetTable.entrySet().iterator(); while (iterator.hasNext()) { Map.Entry<MessageQueue, Long> entry = iterator.next(); System.out.printf("%-40s %-40d %-40d%n", UtilAll.frontStringAtLeast(entry.getKey().getBrokerName(), 32), entry.getKey().getQueueId(), entry.getValue()); } } catch (Exception e) { throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e); } finally { defaultMQAdminExt.shutdown(); } }  @Override String commandName(); @Override String commandDesc(); @Override Options buildCommandlineOptions(Options options); @Override void execute(CommandLine commandLine, Options options, RPCHook rpcHook); }
ResetOffsetByTimeCommand implements SubCommand { @Override public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) throws SubCommandException { DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook); defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis())); try { String group = commandLine.getOptionValue("g").trim(); String topic = commandLine.getOptionValue("t").trim(); String timeStampStr = commandLine.getOptionValue("s").trim(); long timestamp = timeStampStr.equals("now") ? System.currentTimeMillis() : 0; try { if (timestamp == 0) { timestamp = Long.parseLong(timeStampStr); } } catch (NumberFormatException e) { timestamp = UtilAll.parseDate(timeStampStr, UtilAll.YYYY_MM_DD_HH_MM_SS_SSS).getTime(); } boolean force = true; if (commandLine.hasOption('f')) { force = Boolean.valueOf(commandLine.getOptionValue("f").trim()); } boolean isC = false; if (commandLine.hasOption('c')) { isC = true; } defaultMQAdminExt.start(); Map<MessageQueue, Long> offsetTable; try { offsetTable = defaultMQAdminExt.resetOffsetByTimestamp(topic, group, timestamp, force, isC); } catch (MQClientException e) { if (ResponseCode.CONSUMER_NOT_ONLINE == e.getResponseCode()) { ResetOffsetByTimeOldCommand.resetOffset(defaultMQAdminExt, group, topic, timestamp, force, timeStampStr); return; } throw e; } System.out.printf("rollback consumer offset by specified group[%s], topic[%s], force[%s], timestamp(string)[%s], timestamp(long)[%s]%n", group, topic, force, timeStampStr, timestamp); System.out.printf("%-40s %-40s %-40s%n", "#brokerName", "#queueId", "#offset"); Iterator<Map.Entry<MessageQueue, Long>> iterator = offsetTable.entrySet().iterator(); while (iterator.hasNext()) { Map.Entry<MessageQueue, Long> entry = iterator.next(); System.out.printf("%-40s %-40d %-40d%n", UtilAll.frontStringAtLeast(entry.getKey().getBrokerName(), 32), entry.getKey().getQueueId(), entry.getValue()); } } catch (Exception e) { throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e); } finally { defaultMQAdminExt.shutdown(); } }  @Override String commandName(); @Override String commandDesc(); @Override Options buildCommandlineOptions(Options options); @Override void execute(CommandLine commandLine, Options options, RPCHook rpcHook);  }