@Test public void rePartitionTest() throws Exception { int count = 1024; long timeout = 500000L; List<ByteBuffer> messages = MessageUtils.build(count, 1024); long length = messages.stream().mapToInt(Buffer::remaining).sum(); WriteRequest [] writeRequests = IntStream.range(0, messages.size()) .mapToObj(i -> new WriteRequest(partitions[i % partitions.length], messages.get(i))) .toArray(WriteRequest[]::new); final EventFuture<WriteResult> future = new EventFuture<>(); store.asyncWrite(QosLevel.RECEIVE, future,writeRequests); store.commit(store.rightPosition()); long t0 = SystemClock.now(); while (SystemClock.now() - t0 < timeout && store.indexPosition() < length) { Thread.sleep(10L); } store.rePartition(new Short[] {1, 2, 5}); destroyStore(); if (null == bufferPool) { bufferPool = PreloadBufferPool.getInstance(); bufferPool.addPreLoad(128 * 1024 * 1024, 2, 4); bufferPool.addPreLoad(512 * 1024, 2, 4); } PartitionGroupStoreManager.Config config = new PartitionGroupStoreManager.Config(DEFAULT_MAX_MESSAGE_LENGTH, DEFAULT_WRITE_REQUEST_CACHE_SIZE, DEFAULT_FLUSH_INTERVAL_MS, DEFAULT_WRITE_TIMEOUT_MS, DEFAULT_MAX_DIRTY_SIZE, 6000, new PositioningStore.Config(128 * 1024 * 1024), new PositioningStore.Config(512 * 1024)); this.store = new PartitionGroupStoreManager(topic, partitionGroup, groupBase, config, bufferPool); this.store.recover(); this.store.start(); this.store.enable(); }
synchronized void rePartition(Short[] partitions) throws IOException { for (short partition : partitions) { if (!partitionMap.containsKey(partition)) { addPartition(partition); } } List<Short> partitionList = Arrays.asList(partitions); List<Short> toBeRemoved = new ArrayList<>(); for (Map.Entry<Short, Partition> entry : partitionMap.entrySet()) { if (!partitionList.contains(entry.getKey())) { toBeRemoved.add(entry.getKey()); } } for (Short partition : toBeRemoved) { removePartition(partition); } }
PartitionGroupStoreManager extends Service implements ReplicableStore, Closeable { synchronized void rePartition(Short[] partitions) throws IOException { for (short partition : partitions) { if (!partitionMap.containsKey(partition)) { addPartition(partition); } } List<Short> partitionList = Arrays.asList(partitions); List<Short> toBeRemoved = new ArrayList<>(); for (Map.Entry<Short, Partition> entry : partitionMap.entrySet()) { if (!partitionList.contains(entry.getKey())) { toBeRemoved.add(entry.getKey()); } } for (Short partition : toBeRemoved) { removePartition(partition); } } }
PartitionGroupStoreManager extends Service implements ReplicableStore, Closeable { synchronized void rePartition(Short[] partitions) throws IOException { for (short partition : partitions) { if (!partitionMap.containsKey(partition)) { addPartition(partition); } } List<Short> partitionList = Arrays.asList(partitions); List<Short> toBeRemoved = new ArrayList<>(); for (Map.Entry<Short, Partition> entry : partitionMap.entrySet()) { if (!partitionList.contains(entry.getKey())) { toBeRemoved.add(entry.getKey()); } } for (Short partition : toBeRemoved) { removePartition(partition); } } PartitionGroupStoreManager(String topic, int partitionGroup, File base, Config config,
                                      PreloadBufferPool bufferPool); }
PartitionGroupStoreManager extends Service implements ReplicableStore, Closeable { synchronized void rePartition(Short[] partitions) throws IOException { for (short partition : partitions) { if (!partitionMap.containsKey(partition)) { addPartition(partition); } } List<Short> partitionList = Arrays.asList(partitions); List<Short> toBeRemoved = new ArrayList<>(); for (Map.Entry<Short, Partition> entry : partitionMap.entrySet()) { if (!partitionList.contains(entry.getKey())) { toBeRemoved.add(entry.getKey()); } } for (Short partition : toBeRemoved) { removePartition(partition); } } PartitionGroupStoreManager(String topic, int partitionGroup, File base, Config config,
                                      PreloadBufferPool bufferPool); void recover(); String getTopic(); int getPartitionGroup(); ReadResult read(short partition, long index, int count, long maxSize); long getLeftIndex(short partition); long getLeftIndexAndCheck(short partition); long getRightIndex(short partition); long getRightIndexAndCheck(short partition); @Override boolean serviceStatus(); @Override void enable(); @Override void disable(); @Override void setRightPosition(long position); @Override void clear(long position); @Override long commitPosition(); @Override int term(); @Override void term(int term); @Override ByteBuffer readEntryBuffer(long position, int length); @Override long appendEntryBuffer(ByteBuffer byteBuffer); @Override long position(long position, int offsetCount); @Override int lastEntryTerm(); @Override void commit(long position); @Override int getEntryTerm(long position); @Override long leftPosition(); long rightPosition(); @Override void close(); long getIndex(short partition, long timestamp); }
PartitionGroupStoreManager extends Service implements ReplicableStore, Closeable { synchronized void rePartition(Short[] partitions) throws IOException { for (short partition : partitions) { if (!partitionMap.containsKey(partition)) { addPartition(partition); } } List<Short> partitionList = Arrays.asList(partitions); List<Short> toBeRemoved = new ArrayList<>(); for (Map.Entry<Short, Partition> entry : partitionMap.entrySet()) { if (!partitionList.contains(entry.getKey())) { toBeRemoved.add(entry.getKey()); } } for (Short partition : toBeRemoved) { removePartition(partition); } } PartitionGroupStoreManager(String topic, int partitionGroup, File base, Config config,
                                      PreloadBufferPool bufferPool); void recover(); String getTopic(); int getPartitionGroup(); ReadResult read(short partition, long index, int count, long maxSize); long getLeftIndex(short partition); long getLeftIndexAndCheck(short partition); long getRightIndex(short partition); long getRightIndexAndCheck(short partition); @Override boolean serviceStatus(); @Override void enable(); @Override void disable(); @Override void setRightPosition(long position); @Override void clear(long position); @Override long commitPosition(); @Override int term(); @Override void term(int term); @Override ByteBuffer readEntryBuffer(long position, int length); @Override long appendEntryBuffer(ByteBuffer byteBuffer); @Override long position(long position, int offsetCount); @Override int lastEntryTerm(); @Override void commit(long position); @Override int getEntryTerm(long position); @Override long leftPosition(); long rightPosition(); @Override void close(); long getIndex(short partition, long timestamp);  }