@Test public void shouldDecodePreviousVersion() throws Exception { List<TaskId> activeTasks = Arrays.asList(new TaskId(0, 0), new TaskId(0, 0), new TaskId(0, 1), new TaskId(1, 0)); Map<TaskId, Set<TopicPartition>> standbyTasks = new HashMap<>(); standbyTasks.put(new TaskId(1, 1), Utils.mkSet(new TopicPartition("t1", 1), new TopicPartition("t2", 1))); standbyTasks.put(new TaskId(2, 0), Utils.mkSet(new TopicPartition("t3", 0), new TopicPartition("t3", 0))); final AssignmentInfo oldVersion = new AssignmentInfo(1, activeTasks, standbyTasks, null); final AssignmentInfo decoded = AssignmentInfo.decode(encodeV1(oldVersion)); assertEquals(oldVersion.activeTasks, decoded.activeTasks); assertEquals(oldVersion.standbyTasks, decoded.standbyTasks); assertEquals(0, decoded.partitionsByHost.size()); assertEquals(2, decoded.version); }
public static AssignmentInfo decode(ByteBuffer data) { data.rewind(); DataInputStream in = new DataInputStream(new ByteBufferInputStream(data)); try { int version = in.readInt(); if (version < 0 || version > CURRENT_VERSION) { TaskAssignmentException ex = new TaskAssignmentException("Unknown assignment data version: " + version); log.error(ex.getMessage(), ex); throw ex; } int count = in.readInt(); List<TaskId> activeTasks = new ArrayList<>(count); for (int i = 0; i < count; i++) { activeTasks.add(TaskId.readFrom(in)); } count = in.readInt(); Map<TaskId, Set<TopicPartition>> standbyTasks = new HashMap<>(count); for (int i = 0; i < count; i++) { TaskId id = TaskId.readFrom(in); standbyTasks.put(id, readTopicPartitions(in)); } Map<HostInfo, Set<TopicPartition>> hostStateToTopicPartitions = new HashMap<>(); if (version == CURRENT_VERSION) { int numEntries = in.readInt(); for (int i = 0; i < numEntries; i++) { HostInfo hostInfo = new HostInfo(in.readUTF(), in.readInt()); hostStateToTopicPartitions.put(hostInfo, readTopicPartitions(in)); } } return new AssignmentInfo(activeTasks, standbyTasks, hostStateToTopicPartitions); } catch (IOException ex) { throw new TaskAssignmentException("Failed to decode AssignmentInfo", ex); } }
AssignmentInfo { public static AssignmentInfo decode(ByteBuffer data) { data.rewind(); DataInputStream in = new DataInputStream(new ByteBufferInputStream(data)); try { int version = in.readInt(); if (version < 0 || version > CURRENT_VERSION) { TaskAssignmentException ex = new TaskAssignmentException("Unknown assignment data version: " + version); log.error(ex.getMessage(), ex); throw ex; } int count = in.readInt(); List<TaskId> activeTasks = new ArrayList<>(count); for (int i = 0; i < count; i++) { activeTasks.add(TaskId.readFrom(in)); } count = in.readInt(); Map<TaskId, Set<TopicPartition>> standbyTasks = new HashMap<>(count); for (int i = 0; i < count; i++) { TaskId id = TaskId.readFrom(in); standbyTasks.put(id, readTopicPartitions(in)); } Map<HostInfo, Set<TopicPartition>> hostStateToTopicPartitions = new HashMap<>(); if (version == CURRENT_VERSION) { int numEntries = in.readInt(); for (int i = 0; i < numEntries; i++) { HostInfo hostInfo = new HostInfo(in.readUTF(), in.readInt()); hostStateToTopicPartitions.put(hostInfo, readTopicPartitions(in)); } } return new AssignmentInfo(activeTasks, standbyTasks, hostStateToTopicPartitions); } catch (IOException ex) { throw new TaskAssignmentException("Failed to decode AssignmentInfo", ex); } } }
AssignmentInfo { public static AssignmentInfo decode(ByteBuffer data) { data.rewind(); DataInputStream in = new DataInputStream(new ByteBufferInputStream(data)); try { int version = in.readInt(); if (version < 0 || version > CURRENT_VERSION) { TaskAssignmentException ex = new TaskAssignmentException("Unknown assignment data version: " + version); log.error(ex.getMessage(), ex); throw ex; } int count = in.readInt(); List<TaskId> activeTasks = new ArrayList<>(count); for (int i = 0; i < count; i++) { activeTasks.add(TaskId.readFrom(in)); } count = in.readInt(); Map<TaskId, Set<TopicPartition>> standbyTasks = new HashMap<>(count); for (int i = 0; i < count; i++) { TaskId id = TaskId.readFrom(in); standbyTasks.put(id, readTopicPartitions(in)); } Map<HostInfo, Set<TopicPartition>> hostStateToTopicPartitions = new HashMap<>(); if (version == CURRENT_VERSION) { int numEntries = in.readInt(); for (int i = 0; i < numEntries; i++) { HostInfo hostInfo = new HostInfo(in.readUTF(), in.readInt()); hostStateToTopicPartitions.put(hostInfo, readTopicPartitions(in)); } } return new AssignmentInfo(activeTasks, standbyTasks, hostStateToTopicPartitions); } catch (IOException ex) { throw new TaskAssignmentException("Failed to decode AssignmentInfo", ex); } } AssignmentInfo(List<TaskId> activeTasks, Map<TaskId, Set<TopicPartition>> standbyTasks,
                          Map<HostInfo, Set<TopicPartition>> hostState); protected  AssignmentInfo(int version, List<TaskId> activeTasks, Map<TaskId, Set<TopicPartition>> standbyTasks,
                             Map<HostInfo, Set<TopicPartition>> hostState); }
AssignmentInfo { public static AssignmentInfo decode(ByteBuffer data) { data.rewind(); DataInputStream in = new DataInputStream(new ByteBufferInputStream(data)); try { int version = in.readInt(); if (version < 0 || version > CURRENT_VERSION) { TaskAssignmentException ex = new TaskAssignmentException("Unknown assignment data version: " + version); log.error(ex.getMessage(), ex); throw ex; } int count = in.readInt(); List<TaskId> activeTasks = new ArrayList<>(count); for (int i = 0; i < count; i++) { activeTasks.add(TaskId.readFrom(in)); } count = in.readInt(); Map<TaskId, Set<TopicPartition>> standbyTasks = new HashMap<>(count); for (int i = 0; i < count; i++) { TaskId id = TaskId.readFrom(in); standbyTasks.put(id, readTopicPartitions(in)); } Map<HostInfo, Set<TopicPartition>> hostStateToTopicPartitions = new HashMap<>(); if (version == CURRENT_VERSION) { int numEntries = in.readInt(); for (int i = 0; i < numEntries; i++) { HostInfo hostInfo = new HostInfo(in.readUTF(), in.readInt()); hostStateToTopicPartitions.put(hostInfo, readTopicPartitions(in)); } } return new AssignmentInfo(activeTasks, standbyTasks, hostStateToTopicPartitions); } catch (IOException ex) { throw new TaskAssignmentException("Failed to decode AssignmentInfo", ex); } } AssignmentInfo(List<TaskId> activeTasks, Map<TaskId, Set<TopicPartition>> standbyTasks,
                          Map<HostInfo, Set<TopicPartition>> hostState); protected  AssignmentInfo(int version, List<TaskId> activeTasks, Map<TaskId, Set<TopicPartition>> standbyTasks,
                             Map<HostInfo, Set<TopicPartition>> hostState); ByteBuffer encode(); static AssignmentInfo decode(ByteBuffer data); @Override int hashCode(); @Override boolean equals(Object o); @Override String toString(); }
AssignmentInfo { public static AssignmentInfo decode(ByteBuffer data) { data.rewind(); DataInputStream in = new DataInputStream(new ByteBufferInputStream(data)); try { int version = in.readInt(); if (version < 0 || version > CURRENT_VERSION) { TaskAssignmentException ex = new TaskAssignmentException("Unknown assignment data version: " + version); log.error(ex.getMessage(), ex); throw ex; } int count = in.readInt(); List<TaskId> activeTasks = new ArrayList<>(count); for (int i = 0; i < count; i++) { activeTasks.add(TaskId.readFrom(in)); } count = in.readInt(); Map<TaskId, Set<TopicPartition>> standbyTasks = new HashMap<>(count); for (int i = 0; i < count; i++) { TaskId id = TaskId.readFrom(in); standbyTasks.put(id, readTopicPartitions(in)); } Map<HostInfo, Set<TopicPartition>> hostStateToTopicPartitions = new HashMap<>(); if (version == CURRENT_VERSION) { int numEntries = in.readInt(); for (int i = 0; i < numEntries; i++) { HostInfo hostInfo = new HostInfo(in.readUTF(), in.readInt()); hostStateToTopicPartitions.put(hostInfo, readTopicPartitions(in)); } } return new AssignmentInfo(activeTasks, standbyTasks, hostStateToTopicPartitions); } catch (IOException ex) { throw new TaskAssignmentException("Failed to decode AssignmentInfo", ex); } } AssignmentInfo(List<TaskId> activeTasks, Map<TaskId, Set<TopicPartition>> standbyTasks,
                          Map<HostInfo, Set<TopicPartition>> hostState); protected  AssignmentInfo(int version, List<TaskId> activeTasks, Map<TaskId, Set<TopicPartition>> standbyTasks,
                             Map<HostInfo, Set<TopicPartition>> hostState); ByteBuffer encode(); static AssignmentInfo decode(ByteBuffer data); @Override int hashCode(); @Override boolean equals(Object o); @Override String toString(); final int version; final List<TaskId> activeTasks; final Map<TaskId, Set<TopicPartition>> standbyTasks; final Map<HostInfo, Set<TopicPartition>> partitionsByHost; }