BootstrapPullThread extends BasePullThread { protected BackoffTimer getRetriesBeforeCkptCleanup() { return _retriesBeforeCkptCleanup; } BootstrapPullThread(String name,
      DatabusSourcesConnection sourcesConn,
      DbusEventBuffer dbusEventBuffer,
      ConnectionStateFactory connStateFactory,
      Set<ServerInfo> bootstrapServers,
      List<DbusKeyCompositeFilterConfig> bootstrapFilterConfigs,
      double pullerBufferUtilPct,
      MBeanServer mbeanServer,
      DbusEventFactory eventFactory
      );  BootstrapPullThread(String name,
                             DatabusSourcesConnection sourcesConn,
                             DbusEventBuffer dbusEventBuffer,
                             ConnectionStateFactory connStateFactory,
                             Set<ServerInfo> bootstrapServers,
                             List<DbusKeyCompositeFilterConfig> bootstrapFilterConfigs,
                             double pullerBufferUtilPct,
                             MBeanServer mbeanServer,
                             DbusEventFactory eventFactory,
                             ReentrantLock v3BootstrapLock,
                             Logger log);  static final Short START_OF_SNAPSHOT_SRCID; static final Short START_OF_CATCHUP_SRCID; static final Short END_OF_BOOTSTRAP_SRCID; }
@Test public void testTransition_HappyPathWithStartScn() throws Exception { BootstrapPullThread bsPuller = createBootstrapPullThread(false, false, false); Checkpoint cp = _ckptHandlerSource1.createInitialBootstrapCheckpoint(null, 50L); cp.setBootstrapServerInfo(_serverInfoName); cp.setBootstrapStartScn(100L); bsPuller.getComponentStatus().start(); ConnectionState connState = bsPuller.getConnectionState(); connState.switchToBootstrap(cp); testTransitionCase(bsPuller, StateId.BOOTSTRAP, StateId.REQUEST_STREAM, cp); ServerInfo bstServerInfo = ServerInfo.buildServerInfoFromHostPort(cp.getBootstrapServerInfo(), DbusConstants.HOSTPORT_DELIMITER); Assert.assertEquals(bsPuller.getConnectionState().getCurrentBSServerInfo(), bstServerInfo); Assert.assertEquals(cp.getBootstrapStartScn().longValue(), 100L, "Cleared Bootstrap StartSCN"); Assert.assertEquals(cp.getBootstrapSinceScn().longValue(), 50L, "Cleared Bootstrap SinceSCN"); String actualHost = bsPuller.getCurentServer().getAddress().getHostName(); int actualPort = bsPuller.getCurentServer().getAddress().getPort(); Assert.assertEquals(actualHost, _host, "Current Server Host Check"); Assert.assertEquals(actualPort, _port, "Server Port Check"); int numRetries = bsPuller.getRetriesBeforeCkptCleanup().getRemainingRetriesNum(); Assert.assertEquals(numRetries, 1000, "NumRetries Check"); }
@Test public void testTransition_RestartWithNoServerInfo() throws Exception { BootstrapPullThread bsPuller = createBootstrapPullThread(false, false, false); Checkpoint cp = _ckptHandlerSource1.createInitialBootstrapCheckpoint(null, 50L); cp.setBootstrapStartScn(1111L); bsPuller.getComponentStatus().start(); ConnectionState connState = bsPuller.getConnectionState(); connState.switchToBootstrap(cp); testTransitionCase(bsPuller, StateId.BOOTSTRAP, StateId.REQUEST_START_SCN, cp); Assert.assertEquals(cp.getBootstrapStartScn().longValue(), Checkpoint.UNSET_BOOTSTRAP_START_SCN, "Cleared Bootstrap StartSCN"); Assert.assertEquals(cp.getBootstrapSinceScn().longValue(), 50L, "Cleared Bootstrap SinceSCN"); int numRetries = bsPuller.getRetriesBeforeCkptCleanup().getRemainingRetriesNum(); Assert.assertEquals(numRetries, 1000, "NumRetries Check"); }
@Test public void testTransition_RestartDueToServerInfoErrors() throws Exception { final String dummyHost = "NonExistantHost"; final String dummyServerInfoName = dummyHost + ":" + _port; BootstrapPullThread bsPuller = createBootstrapPullThread(false, false, false); Checkpoint cp = _ckptHandlerSource1.createInitialBootstrapCheckpoint(null, 50L); cp.setBootstrapStartScn(1111L); cp.setBootstrapServerInfo(dummyServerInfoName); bsPuller.getComponentStatus().start(); ConnectionState connState = bsPuller.getConnectionState(); connState.switchToBootstrap(cp); testTransitionCase(bsPuller, StateId.BOOTSTRAP, StateId.REQUEST_START_SCN, cp); Assert.assertEquals(cp.getBootstrapStartScn().longValue(), Checkpoint.UNSET_BOOTSTRAP_START_SCN, "Cleared Bootstrap StartSCN"); Assert.assertEquals(cp.getBootstrapSinceScn().longValue(), 50L, "Cleared Bootstrap SinceSCN"); int numRetries = bsPuller.getRetriesBeforeCkptCleanup().getRemainingRetriesNum(); Assert.assertEquals(numRetries, 1000, "NumRetries Check"); }
@Test public void testTransition_RestartDueToMalformedServerInfo() throws Exception { final String dummyHost = "NonExistantHost"; final String malformedServerInfoName = dummyHost + _port; BootstrapPullThread bsPuller = createBootstrapPullThread(false, false, false); Checkpoint cp = _ckptHandlerSource1.createInitialBootstrapCheckpoint(null, 50L); cp.setBootstrapStartScn(1111L); cp.setBootstrapServerInfo(malformedServerInfoName); bsPuller.getComponentStatus().start(); ConnectionState connState = bsPuller.getConnectionState(); connState.switchToBootstrap(cp); testTransitionCase(bsPuller, StateId.BOOTSTRAP, StateId.REQUEST_START_SCN, cp); Assert.assertEquals(cp.getBootstrapStartScn().longValue(), Checkpoint.UNSET_BOOTSTRAP_START_SCN, "Cleared Bootstrap StartSCN"); Assert.assertEquals(cp.getBootstrapSinceScn().longValue(), 50L, "Cleared Bootstrap SinceSCN"); int numRetries = bsPuller.getRetriesBeforeCkptCleanup().getRemainingRetriesNum(); Assert.assertEquals(numRetries, 1000, "NumRetries Check"); }