@SuppressWarnings("unchecked") @Test public void testFallback() throws InjectionException, NetworkException { final Codec codec = Tang.Factory.getTang().newInjector().getInstance(SerializableCodec.class); final TableAccessReqMsg getMsg = TableAccessReqMsg.newBuilder() .setOrigId(NON_OWNER_EXECUTOR_ID) .setOpType(OpType.GET) .setTableId(TABLE_ID) .setDataKey( DataKey.newBuilder() .setKey(ByteBuffer.wrap(codec.encode(KEY))) .build()) .setDataValue(null) .build(); final TableAccessMsg tableAccessMsg = TableAccessMsg.newBuilder() .setType(TableAccessMsgType.TableAccessReqMsg) .setOperationId(OP_ID) .setTableAccessReqMsg(getMsg) .build(); fallbackManager.onTableAccessReqMessage(NON_OWNER_EXECUTOR_ID, tableAccessMsg); doAnswer(invocation -> { final Object[] arguments = invocation.getArguments(); final String destId = (String) arguments[0]; final long opId = (long) arguments[1]; final TableAccessReqMsg tableAccessReqMsg = (TableAccessReqMsg) arguments[2]; assertEquals("The original executor is different", NON_OWNER_EXECUTOR_ID, tableAccessReqMsg.getOrigId()); assertEquals("The destination is different", OWNER_EXECUTOR_ID, destId); assertEquals("The op id is different", OP_ID, opId); assertEquals("The table id is different", TABLE_ID, tableAccessReqMsg.getTableId()); return null; }).when(msgSender).sendTableAccessReqMsg(anyString(), anyLong(), any(TableAccessReqMsg.class)); }
@SuppressWarnings("unchecked") void onTableAccessReqMessage(final String srcId, final TableAccessMsg tableAccessMsg) { final long opId = tableAccessMsg.getOperationId(); final TableAccessReqMsg innerMsg = tableAccessMsg.getTableAccessReqMsg(); final String tableId = innerMsg.getTableId(); LOG.log(Level.INFO, "The table access request (Table: {0}, opId: {1}, origId: {2}) has failed at {3}." + " Will try redirection.", new Object[] {tableId, opId, innerMsg.getOrigId(), srcId}); try { final AllocatedTable table = tableManagerFuture.get().getAllocatedTable(tableId); final Injector injector = Tang.Factory.getTang().newInjector(table.getTableConfiguration().getConfiguration()); final StreamingCodec keyCodec = injector.getNamedInstance(KeyCodec.class); final BlockPartitioner blockPartitioner = injector.getInstance(BlockPartitioner.class); final int blockId = blockPartitioner.getBlockId(keyCodec.decode(innerMsg.getDataKey().getKey().array())); while (true) { final String executorIdToSendMsg = table.getOwnerId(blockId); LOG.log(Level.INFO, "The table access request (Table: {0}, opId: {1}, origId: {2}) will be redirected to {3}.", new Object[] {tableId, opId, innerMsg.getOrigId(), executorIdToSendMsg}); try { messageSenderFuture.get().sendTableAccessReqMsg(executorIdToSendMsg, opId, innerMsg); break; } catch (NetworkException e) { LOG.log(Level.WARNING, "NetworkException while sending a msg. Resend", e); } LOG.log(Level.INFO, "Wait {0} ms before resending a msg", RESEND_INTERVAL_MS); try { Thread.sleep(RESEND_INTERVAL_MS); } catch (final InterruptedException e) { LOG.log(Level.FINEST, "Interrupted while waiting for ownership cache to be updated", e); } } } catch (TableNotExistException | InjectionException e) { LOG.log(Level.INFO, "Redirection of the table access request has failed.", e); throw new RuntimeException(e); } }
FallbackManager { @SuppressWarnings("unchecked") void onTableAccessReqMessage(final String srcId, final TableAccessMsg tableAccessMsg) { final long opId = tableAccessMsg.getOperationId(); final TableAccessReqMsg innerMsg = tableAccessMsg.getTableAccessReqMsg(); final String tableId = innerMsg.getTableId(); LOG.log(Level.INFO, "The table access request (Table: {0}, opId: {1}, origId: {2}) has failed at {3}." + " Will try redirection.", new Object[] {tableId, opId, innerMsg.getOrigId(), srcId}); try { final AllocatedTable table = tableManagerFuture.get().getAllocatedTable(tableId); final Injector injector = Tang.Factory.getTang().newInjector(table.getTableConfiguration().getConfiguration()); final StreamingCodec keyCodec = injector.getNamedInstance(KeyCodec.class); final BlockPartitioner blockPartitioner = injector.getInstance(BlockPartitioner.class); final int blockId = blockPartitioner.getBlockId(keyCodec.decode(innerMsg.getDataKey().getKey().array())); while (true) { final String executorIdToSendMsg = table.getOwnerId(blockId); LOG.log(Level.INFO, "The table access request (Table: {0}, opId: {1}, origId: {2}) will be redirected to {3}.", new Object[] {tableId, opId, innerMsg.getOrigId(), executorIdToSendMsg}); try { messageSenderFuture.get().sendTableAccessReqMsg(executorIdToSendMsg, opId, innerMsg); break; } catch (NetworkException e) { LOG.log(Level.WARNING, "NetworkException while sending a msg. Resend", e); } LOG.log(Level.INFO, "Wait {0} ms before resending a msg", RESEND_INTERVAL_MS); try { Thread.sleep(RESEND_INTERVAL_MS); } catch (final InterruptedException e) { LOG.log(Level.FINEST, "Interrupted while waiting for ownership cache to be updated", e); } } } catch (TableNotExistException | InjectionException e) { LOG.log(Level.INFO, "Redirection of the table access request has failed.", e); throw new RuntimeException(e); } } }
FallbackManager { @SuppressWarnings("unchecked") void onTableAccessReqMessage(final String srcId, final TableAccessMsg tableAccessMsg) { final long opId = tableAccessMsg.getOperationId(); final TableAccessReqMsg innerMsg = tableAccessMsg.getTableAccessReqMsg(); final String tableId = innerMsg.getTableId(); LOG.log(Level.INFO, "The table access request (Table: {0}, opId: {1}, origId: {2}) has failed at {3}." + " Will try redirection.", new Object[] {tableId, opId, innerMsg.getOrigId(), srcId}); try { final AllocatedTable table = tableManagerFuture.get().getAllocatedTable(tableId); final Injector injector = Tang.Factory.getTang().newInjector(table.getTableConfiguration().getConfiguration()); final StreamingCodec keyCodec = injector.getNamedInstance(KeyCodec.class); final BlockPartitioner blockPartitioner = injector.getInstance(BlockPartitioner.class); final int blockId = blockPartitioner.getBlockId(keyCodec.decode(innerMsg.getDataKey().getKey().array())); while (true) { final String executorIdToSendMsg = table.getOwnerId(blockId); LOG.log(Level.INFO, "The table access request (Table: {0}, opId: {1}, origId: {2}) will be redirected to {3}.", new Object[] {tableId, opId, innerMsg.getOrigId(), executorIdToSendMsg}); try { messageSenderFuture.get().sendTableAccessReqMsg(executorIdToSendMsg, opId, innerMsg); break; } catch (NetworkException e) { LOG.log(Level.WARNING, "NetworkException while sending a msg. Resend", e); } LOG.log(Level.INFO, "Wait {0} ms before resending a msg", RESEND_INTERVAL_MS); try { Thread.sleep(RESEND_INTERVAL_MS); } catch (final InterruptedException e) { LOG.log(Level.FINEST, "Interrupted while waiting for ownership cache to be updated", e); } } } catch (TableNotExistException | InjectionException e) { LOG.log(Level.INFO, "Redirection of the table access request has failed.", e); throw new RuntimeException(e); } } @Inject private  FallbackManager(final InjectionFuture<MessageSender> messageSenderFuture,
                          final InjectionFuture<TableManager> tableManagerFuture); }
FallbackManager { @SuppressWarnings("unchecked") void onTableAccessReqMessage(final String srcId, final TableAccessMsg tableAccessMsg) { final long opId = tableAccessMsg.getOperationId(); final TableAccessReqMsg innerMsg = tableAccessMsg.getTableAccessReqMsg(); final String tableId = innerMsg.getTableId(); LOG.log(Level.INFO, "The table access request (Table: {0}, opId: {1}, origId: {2}) has failed at {3}." + " Will try redirection.", new Object[] {tableId, opId, innerMsg.getOrigId(), srcId}); try { final AllocatedTable table = tableManagerFuture.get().getAllocatedTable(tableId); final Injector injector = Tang.Factory.getTang().newInjector(table.getTableConfiguration().getConfiguration()); final StreamingCodec keyCodec = injector.getNamedInstance(KeyCodec.class); final BlockPartitioner blockPartitioner = injector.getInstance(BlockPartitioner.class); final int blockId = blockPartitioner.getBlockId(keyCodec.decode(innerMsg.getDataKey().getKey().array())); while (true) { final String executorIdToSendMsg = table.getOwnerId(blockId); LOG.log(Level.INFO, "The table access request (Table: {0}, opId: {1}, origId: {2}) will be redirected to {3}.", new Object[] {tableId, opId, innerMsg.getOrigId(), executorIdToSendMsg}); try { messageSenderFuture.get().sendTableAccessReqMsg(executorIdToSendMsg, opId, innerMsg); break; } catch (NetworkException e) { LOG.log(Level.WARNING, "NetworkException while sending a msg. Resend", e); } LOG.log(Level.INFO, "Wait {0} ms before resending a msg", RESEND_INTERVAL_MS); try { Thread.sleep(RESEND_INTERVAL_MS); } catch (final InterruptedException e) { LOG.log(Level.FINEST, "Interrupted while waiting for ownership cache to be updated", e); } } } catch (TableNotExistException | InjectionException e) { LOG.log(Level.INFO, "Redirection of the table access request has failed.", e); throw new RuntimeException(e); } } @Inject private  FallbackManager(final InjectionFuture<MessageSender> messageSenderFuture,
                          final InjectionFuture<TableManager> tableManagerFuture);  }
FallbackManager { @SuppressWarnings("unchecked") void onTableAccessReqMessage(final String srcId, final TableAccessMsg tableAccessMsg) { final long opId = tableAccessMsg.getOperationId(); final TableAccessReqMsg innerMsg = tableAccessMsg.getTableAccessReqMsg(); final String tableId = innerMsg.getTableId(); LOG.log(Level.INFO, "The table access request (Table: {0}, opId: {1}, origId: {2}) has failed at {3}." + " Will try redirection.", new Object[] {tableId, opId, innerMsg.getOrigId(), srcId}); try { final AllocatedTable table = tableManagerFuture.get().getAllocatedTable(tableId); final Injector injector = Tang.Factory.getTang().newInjector(table.getTableConfiguration().getConfiguration()); final StreamingCodec keyCodec = injector.getNamedInstance(KeyCodec.class); final BlockPartitioner blockPartitioner = injector.getInstance(BlockPartitioner.class); final int blockId = blockPartitioner.getBlockId(keyCodec.decode(innerMsg.getDataKey().getKey().array())); while (true) { final String executorIdToSendMsg = table.getOwnerId(blockId); LOG.log(Level.INFO, "The table access request (Table: {0}, opId: {1}, origId: {2}) will be redirected to {3}.", new Object[] {tableId, opId, innerMsg.getOrigId(), executorIdToSendMsg}); try { messageSenderFuture.get().sendTableAccessReqMsg(executorIdToSendMsg, opId, innerMsg); break; } catch (NetworkException e) { LOG.log(Level.WARNING, "NetworkException while sending a msg. Resend", e); } LOG.log(Level.INFO, "Wait {0} ms before resending a msg", RESEND_INTERVAL_MS); try { Thread.sleep(RESEND_INTERVAL_MS); } catch (final InterruptedException e) { LOG.log(Level.FINEST, "Interrupted while waiting for ownership cache to be updated", e); } } } catch (TableNotExistException | InjectionException e) { LOG.log(Level.INFO, "Redirection of the table access request has failed.", e); throw new RuntimeException(e); } } @Inject private  FallbackManager(final InjectionFuture<MessageSender> messageSenderFuture,
                          final InjectionFuture<TableManager> tableManagerFuture);   }