BrokerCfg { public ClusterCfg getCluster() { return cluster; }  void init(final String brokerBase); void init(final String brokerBase, final Environment environment); NetworkCfg getNetwork(); void setNetwork(final NetworkCfg network); ClusterCfg getCluster(); void setCluster(final ClusterCfg cluster); ThreadsCfg getThreads(); void setThreads(final ThreadsCfg threads); DataCfg getData(); void setData(final DataCfg logs); Map<String, ExporterCfg> getExporters(); void setExporters(final Map<String, ExporterCfg> exporters); EmbeddedGatewayCfg getGateway(); BrokerCfg setGateway(final EmbeddedGatewayCfg gateway); BackpressureCfg getBackpressure(); BrokerCfg setBackpressure(final BackpressureCfg backpressure); Duration getStepTimeout(); void setStepTimeout(final Duration stepTimeout); boolean isExecutionMetricsExporterEnabled(); void setExecutionMetricsExporterEnabled(final boolean executionMetricsExporterEnabled); @Override String toString(); String toJson();  }
@Test public void shouldReadSpecificSystemClusterConfiguration() { final BrokerCfg cfg = TestConfigReader.readConfig("cluster-cfg", environment); final ClusterCfg cfgCluster = cfg.getCluster(); assertThat(cfgCluster.getInitialContactPoints()).isEmpty(); assertThat(cfgCluster.getNodeId()).isEqualTo(2); assertThat(cfgCluster.getPartitionsCount()).isEqualTo(3); assertThat(cfgCluster.getReplicationFactor()).isEqualTo(4); assertThat(cfgCluster.getClusterSize()).isEqualTo(5); }
@Test public void shouldCreatePartitionIds() { final BrokerCfg cfg = TestConfigReader.readConfig("cluster-cfg", environment); final ClusterCfg cfgCluster = cfg.getCluster(); assertThat(cfgCluster.getPartitionsCount()).isEqualTo(3); final List<Integer> partitionIds = cfgCluster.getPartitionIds(); final int startId = START_PARTITION_ID; assertThat(partitionIds).contains(startId, startId + 1, startId + 2); }
@Test public void shouldOverrideReplicationFactorViaEnvironment() { environment.put(ZEEBE_BROKER_CLUSTER_REPLICATION_FACTOR, "2"); final BrokerCfg cfg = TestConfigReader.readConfig("cluster-cfg", environment); final ClusterCfg cfgCluster = cfg.getCluster(); assertThat(cfgCluster.getReplicationFactor()).isEqualTo(2); }
@Test public void shouldOverridePartitionsCountViaEnvironment() { environment.put(ZEEBE_BROKER_CLUSTER_PARTITIONS_COUNT, "2"); final BrokerCfg cfg = TestConfigReader.readConfig("cluster-cfg", environment); final ClusterCfg cfgCluster = cfg.getCluster(); assertThat(cfgCluster.getPartitionsCount()).isEqualTo(2); }
@Test public void shouldOverrideClusterSizeViaEnvironment() { environment.put(ZEEBE_BROKER_CLUSTER_CLUSTER_SIZE, "2"); final BrokerCfg cfg = TestConfigReader.readConfig("cluster-cfg", environment); final ClusterCfg cfgCluster = cfg.getCluster(); assertThat(cfgCluster.getClusterSize()).isEqualTo(2); }
@Test public void shouldOverrideAllClusterPropertiesViaEnvironment() { environment.put(ZEEBE_BROKER_CLUSTER_CLUSTER_SIZE, "1"); environment.put(ZEEBE_BROKER_CLUSTER_PARTITIONS_COUNT, "2"); environment.put(ZEEBE_BROKER_CLUSTER_REPLICATION_FACTOR, "3"); environment.put(ZEEBE_BROKER_CLUSTER_NODE_ID, "4"); final BrokerCfg cfg = TestConfigReader.readConfig("cluster-cfg", environment); final ClusterCfg cfgCluster = cfg.getCluster(); assertThat(cfgCluster.getClusterSize()).isEqualTo(1); assertThat(cfgCluster.getPartitionsCount()).isEqualTo(2); assertThat(cfgCluster.getReplicationFactor()).isEqualTo(3); assertThat(cfgCluster.getNodeId()).isEqualTo(4); }
@Test public void shouldSetCustomMembershipConfig() { final BrokerCfg brokerCfg = TestConfigReader.readConfig("membership-cfg", environment); final var membershipCfg = brokerCfg.getCluster().getMembership(); assertThat(membershipCfg.isBroadcastDisputes()).isFalse(); assertThat(membershipCfg.isBroadcastUpdates()).isTrue(); assertThat(membershipCfg.isNotifySuspect()).isTrue(); assertThat(membershipCfg.getGossipInterval()).isEqualTo(Duration.ofSeconds(2)); assertThat(membershipCfg.getGossipFanout()).isEqualTo(3); assertThat(membershipCfg.getProbeInterval()).isEqualTo(Duration.ofSeconds(3)); assertThat(membershipCfg.getProbeTimeout()).isEqualTo(Duration.ofSeconds(5)); assertThat(membershipCfg.getSuspectProbes()).isEqualTo(5); assertThat(membershipCfg.getFailureTimeout()).isEqualTo(Duration.ofSeconds(20)); assertThat(membershipCfg.getSyncInterval()).isEqualTo(Duration.ofSeconds(25)); }