SystemContext { private void initSystemContext(final ActorClock clock, final String basePath) { LOG.debug("Initializing system with base path {}", basePath); brokerCfg.init(basePath); validateConfiguration(); stepTimeout = brokerCfg.getStepTimeout(); final var cluster = brokerCfg.getCluster(); final String brokerId = String.format("Broker-%d", cluster.getNodeId()); diagnosticContext = Collections.singletonMap(BROKER_ID_LOG_PROPERTY, brokerId); scheduler = initScheduler(clock, brokerId); setStepTimeout(stepTimeout); } SystemContext(final BrokerCfg brokerCfg, final String basePath, final ActorClock clock); ActorScheduler getScheduler(); BrokerCfg getBrokerConfiguration(); Map<String, String> getDiagnosticContext(); Duration getStepTimeout(); static final Logger LOG; }
@Test public void shouldThrowExceptionIfNodeIdIsNegative() { final BrokerCfg brokerCfg = new BrokerCfg(); brokerCfg.getCluster().setNodeId(-1); expectedException.expect(IllegalArgumentException.class); expectedException.expectMessage( "Node id -1 needs to be non negative and smaller then cluster size 1."); initSystemContext(brokerCfg); }
@Test public void shouldThrowExceptionIfNodeIdIsLargerThenClusterSize() { final BrokerCfg brokerCfg = new BrokerCfg(); brokerCfg.getCluster().setNodeId(2); expectedException.expect(IllegalArgumentException.class); expectedException.expectMessage( "Node id 2 needs to be non negative and smaller then cluster size 1."); initSystemContext(brokerCfg); }
@Test public void shouldThrowExceptionIfReplicationFactorIsNegative() { final BrokerCfg brokerCfg = new BrokerCfg(); brokerCfg.getCluster().setReplicationFactor(-1); expectedException.expect(IllegalArgumentException.class); expectedException.expectMessage( "Replication factor -1 needs to be larger then zero and not larger then cluster size 1."); initSystemContext(brokerCfg); }
@Test public void shouldThrowExceptionIfReplicationFactorIsLargerThenClusterSize() { final BrokerCfg brokerCfg = new BrokerCfg(); brokerCfg.getCluster().setReplicationFactor(2); expectedException.expect(IllegalArgumentException.class); expectedException.expectMessage( "Replication factor 2 needs to be larger then zero and not larger then cluster size 1."); initSystemContext(brokerCfg); }
@Test public void shouldThrowExceptionIfPartitionsCountIsNegative() { final BrokerCfg brokerCfg = new BrokerCfg(); brokerCfg.getCluster().setPartitionsCount(-1); expectedException.expect(IllegalArgumentException.class); expectedException.expectMessage("Partition count must not be smaller then 1."); initSystemContext(brokerCfg); }
@Test public void shouldThrowExceptionIfSnapshotPeriodIsNegative() { final BrokerCfg brokerCfg = new BrokerCfg(); brokerCfg.getData().setSnapshotPeriod(Duration.ofMinutes(-1)); expectedException.expect(IllegalArgumentException.class); expectedException.expectMessage( "Snapshot period PT-1M needs to be larger then or equals to one minute."); initSystemContext(brokerCfg); }
@Test public void shouldThrowExceptionIfSnapshotPeriodIsTooSmall() { final BrokerCfg brokerCfg = new BrokerCfg(); brokerCfg.getData().setSnapshotPeriod(Duration.ofSeconds(1)); expectedException.expect(IllegalArgumentException.class); expectedException.expectMessage( "Snapshot period PT1S needs to be larger then or equals to one minute."); initSystemContext(brokerCfg); }
@Test public void shouldInvalidateConfigIfUseMmapTrueWithReplication() { final BrokerCfg brokerCfg = new BrokerCfg(); brokerCfg.getData().setUseMmap(true); brokerCfg.getCluster().setClusterSize(2); brokerCfg.getCluster().setReplicationFactor(2); expectedException.expect(IllegalStateException.class); expectedException.expectMessage("memory mapped"); initSystemContext(brokerCfg); }