forked from apache/pulsar
/
ServiceConfiguration.java
2336 lines (2110 loc) · 99.6 KB
/
ServiceConfiguration.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.apache.pulsar.broker;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import io.netty.util.internal.PlatformDependent;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Optional;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import lombok.Getter;
import lombok.Setter;
import org.apache.bookkeeper.client.api.DigestType;
import org.apache.pulsar.broker.authorization.PulsarAuthorizationProvider;
import org.apache.pulsar.common.configuration.Category;
import org.apache.pulsar.common.configuration.FieldContext;
import org.apache.pulsar.common.configuration.PulsarConfiguration;
import org.apache.pulsar.common.nar.NarClassLoader;
import org.apache.pulsar.common.policies.data.BacklogQuota;
import org.apache.pulsar.common.policies.data.InactiveTopicDeleteMode;
import org.apache.pulsar.common.policies.data.OffloadedReadPriority;
import org.apache.pulsar.common.policies.data.TopicType;
import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.sasl.SaslConstants;
/**
* Pulsar service configuration object.
*/
@Getter
@Setter
public class ServiceConfiguration implements PulsarConfiguration {
@Category
private static final String CATEGORY_SERVER = "Server";
@Category
private static final String CATEGORY_PROTOCOLS = "Protocols";
@Category
private static final String CATEGORY_STORAGE_BK = "Storage (BookKeeper)";
@Category
private static final String CATEGORY_STORAGE_ML = "Storage (Managed Ledger)";
@Category
private static final String CATEGORY_STORAGE_OFFLOADING = "Storage (Ledger Offloading)";
@Category
private static final String CATEGORY_POLICIES = "Policies";
@Category
private static final String CATEGORY_WEBSOCKET = "WebSocket";
@Category
private static final String CATEGORY_SCHEMA = "Schema";
@Category
private static final String CATEGORY_METRICS = "Metrics";
@Category
private static final String CATEGORY_REPLICATION = "Replication";
@Category
private static final String CATEGORY_LOAD_BALANCER = "Load Balancer";
@Category
private static final String CATEGORY_FUNCTIONS = "Functions";
@Category
private static final String CATEGORY_TLS = "TLS";
@Category
private static final String CATEGORY_KEYSTORE_TLS = "KeyStoreTLS";
@Category
private static final String CATEGORY_AUTHENTICATION = "Authentication";
@Category
private static final String CATEGORY_AUTHORIZATION = "Authorization";
@Category
private static final String CATEGORY_TOKEN_AUTH = "Token Authentication Provider";
@Category
private static final String CATEGORY_SASL_AUTH = "SASL Authentication Provider";
@Category
private static final String CATEGORY_HTTP = "HTTP";
@Category
private static final String CATEGORY_TRANSACTION = "Transaction";
@Category
private static final String CATEGORY_PACKAGES_MANAGEMENT = "Packages Management";
/***** --- pulsar configuration --- ****/
@FieldContext(
category = CATEGORY_SERVER,
required = true,
doc = "The Zookeeper quorum connection string (as a comma-separated list)"
)
private String zookeeperServers;
@Deprecated
@FieldContext(
category = CATEGORY_SERVER,
required = false,
deprecated = true,
doc = "Global Zookeeper quorum connection string (as a comma-separated list)."
+ " Deprecated in favor of using `configurationStoreServers`"
)
private String globalZookeeperServers;
@FieldContext(
category = CATEGORY_SERVER,
required = false,
doc = "Configuration store connection string (as a comma-separated list)"
)
private String configurationStoreServers;
@FieldContext(
category = CATEGORY_SERVER,
doc = "The port for serving binary protobuf requests"
)
private Optional<Integer> brokerServicePort = Optional.of(6650);
@FieldContext(
category = CATEGORY_SERVER,
doc = "The port for serving tls secured binary protobuf requests"
)
private Optional<Integer> brokerServicePortTls = Optional.empty();
@FieldContext(
category = CATEGORY_SERVER,
doc = "The port for serving http requests"
)
private Optional<Integer> webServicePort = Optional.of(8080);
@FieldContext(
category = CATEGORY_SERVER,
doc = "The port for serving https requests"
)
private Optional<Integer> webServicePortTls = Optional.empty();
@FieldContext(
category = CATEGORY_SERVER,
doc = "Hostname or IP address the service binds on"
)
private String bindAddress = "0.0.0.0";
@FieldContext(
category = CATEGORY_SERVER,
doc = "Hostname or IP address the service advertises to the outside world."
+ " If not set, the value of `InetAddress.getLocalHost().getHostname()` is used."
)
private String advertisedAddress;
@FieldContext(category=CATEGORY_SERVER,
doc = "Used to specify multiple advertised listeners for the broker."
+ " The value must format as <listener_name>:pulsar://<host>:<port>,"
+ "multiple listeners should separate with commas."
+ "Do not use this configuration with advertisedAddress and brokerServicePort."
+ "The Default value is absent means use advertisedAddress and brokerServicePort.")
private String advertisedListeners;
@FieldContext(category=CATEGORY_SERVER,
doc = "Used to specify the internal listener name for the broker."
+ "The listener name must contain in the advertisedListeners."
+ "The Default value is absent, the broker uses the first listener as the internal listener.")
private String internalListenerName;
@FieldContext(category=CATEGORY_SERVER,
doc = "Enable or disable the proxy protocol.")
private boolean haProxyProtocolEnabled;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Number of threads to use for Netty Acceptor."
+ " Default is set to `1`"
)
private int numAcceptorThreads = 1;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Number of threads to use for Netty IO."
+ " Default is set to `2 * Runtime.getRuntime().availableProcessors()`"
)
private int numIOThreads = 2 * Runtime.getRuntime().availableProcessors();
@FieldContext(
category = CATEGORY_SERVER,
doc = "Number of threads to use for orderedExecutor."
+ " The ordered executor is used to operate with zookeeper, such as init zookeeper client,"
+ " get namespace policies from zookeeper etc. It also used to split bundle. Default is 8"
)
private int numOrderedExecutorThreads = 8;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Number of threads to use for HTTP requests processing"
+ " Default is set to `2 * Runtime.getRuntime().availableProcessors()`"
)
// Use at least 8 threads to avoid having Jetty go into threads starving and
// having the possibility of getting into a deadlock where a Jetty thread is
// waiting for another HTTP call to complete in same thread.
private int numHttpServerThreads = Math.max(8, 2 * Runtime.getRuntime().availableProcessors());
@FieldContext(
category = CATEGORY_SERVER,
doc = "Number of threads to use for pulsar broker service."
+ " The executor in thread pool will do basic broker operation like load/unload bundle,"
+ " update managedLedgerConfig, update topic/subscription/replicator message dispatch rate,"
+ " do leader election etc. Default is set to 20 "
)
private int numExecutorThreadPoolSize = Runtime.getRuntime().availableProcessors();
@FieldContext(
category = CATEGORY_SERVER,
doc = "Number of thread pool size to use for pulsar zookeeper callback service."
+ "The cache executor thread pool is used for restarting global zookeeper session. "
+ "Default is 10"
)
private int numCacheExecutorThreadPoolSize = 10;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Option to enable busy-wait settings. Default is false. "
+ "WARNING: This option will enable spin-waiting on executors and IO threads in order "
+ "to reduce latency during context switches. The spinning will consume 100% CPU even "
+ "when the broker is not doing any work. It is recommended to reduce the number of IO threads "
+ "and BK client threads to only have few CPU cores busy."
)
private boolean enableBusyWait = false;
@FieldContext(category = CATEGORY_SERVER, doc = "Max concurrent web requests")
private int maxConcurrentHttpRequests = 1024;
@FieldContext(category = CATEGORY_SERVER, doc = "Whether to enable the delayed delivery for messages.")
private boolean delayedDeliveryEnabled = true;
@FieldContext(category = CATEGORY_SERVER, doc = "Class name of the factory that implements the delayed deliver tracker")
private String delayedDeliveryTrackerFactoryClassName = "org.apache.pulsar.broker.delayed.InMemoryDelayedDeliveryTrackerFactory";
@FieldContext(category = CATEGORY_SERVER, doc = "Control the tick time for when retrying on delayed delivery, "
+ " affecting the accuracy of the delivery time compared to the scheduled time. Default is 1 second.")
private long delayedDeliveryTickTimeMillis = 1000;
@FieldContext(category = CATEGORY_SERVER, doc = "Whether to enable the acknowledge of batch local index")
private boolean acknowledgmentAtBatchIndexLevelEnabled = false;
@FieldContext(
category = CATEGORY_WEBSOCKET,
doc = "Enable the WebSocket API service in broker"
)
private boolean webSocketServiceEnabled = false;
@FieldContext(
category = CATEGORY_WEBSOCKET,
doc = "Flag indicates whether to run broker in standalone mode"
)
private boolean isRunningStandalone = false;
@FieldContext(
category = CATEGORY_SERVER,
required = true,
doc = "Name of the cluster to which this broker belongs to"
)
private String clusterName;
@FieldContext(
category = CATEGORY_SERVER,
dynamic = true,
doc = "The maximum number of tenants that each pulsar cluster can create."
+ "This configuration is not precise control, in a concurrent scenario, the threshold will be exceeded."
)
private int maxTenants = 0;
@FieldContext(
category = CATEGORY_SERVER,
dynamic = true,
doc = "Enable cluster's failure-domain which can distribute brokers into logical region"
)
private boolean failureDomainsEnabled = false;
@FieldContext(
category = CATEGORY_SERVER,
doc = "ZooKeeper session timeout in milliseconds"
)
private long zooKeeperSessionTimeoutMillis = 30000;
@FieldContext(
category = CATEGORY_SERVER,
doc = "ZooKeeper operation timeout in seconds"
)
private int zooKeeperOperationTimeoutSeconds = 30;
@FieldContext(
category = CATEGORY_SERVER,
doc = "ZooKeeper cache expiry time in seconds"
)
private int zooKeeperCacheExpirySeconds = 300;
@FieldContext(
category = CATEGORY_SERVER,
dynamic = true,
doc = "Time to wait for broker graceful shutdown. After this time elapses, the process will be killed"
)
private long brokerShutdownTimeoutMs = 60000;
@FieldContext(
category = CATEGORY_SERVER,
dynamic = true,
doc = "Flag to skip broker shutdown when broker handles Out of memory error"
)
private boolean skipBrokerShutdownOnOOM = false;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Amount of seconds to timeout when loading a topic. In situations with many geo-replicated clusters, this may need raised."
)
private long topicLoadTimeoutSeconds = 60;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "Enable backlog quota check. Enforces actions on topic when the quota is reached"
)
private boolean backlogQuotaCheckEnabled = true;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "Whether to enable precise time based backlog quota check. "
+ "Enabling precise time based backlog quota check will cause broker to read first entry in backlog "
+ "of the slowest cursor on a ledger which will mostly result in reading entry from BookKeeper's " +
"disk which can have negative impact on overall performance. "
+ "Disabling precise time based backlog quota check will just use the timestamp indicating when a "
+ "ledger was closed, which is of coarser granularity."
)
private boolean preciseTimeBasedBacklogQuotaCheck = false;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "How often to check for topics that have reached the quota."
+ " It only takes effects when `backlogQuotaCheckEnabled` is true"
)
private int backlogQuotaCheckIntervalInSeconds = 60;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "Default per-topic backlog quota limit by size, less than 0 means no limitation. default is -1."
+ " Increase it if you want to allow larger msg backlog"
)
private long backlogQuotaDefaultLimitGB = -1;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "Default per-topic backlog quota limit by time in second, less than 0 means no limitation. " +
"default is -1. Increase it if you want to allow larger msg backlog"
)
private int backlogQuotaDefaultLimitSecond = -1;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "Default backlog quota retention policy. Default is producer_request_hold\n\n"
+ "'producer_request_hold' Policy which holds producer's send request until the"
+ "resource becomes available (or holding times out)\n"
+ "'producer_exception' Policy which throws javax.jms.ResourceAllocationException to the producer\n"
+ "'consumer_backlog_eviction' Policy which evicts the oldest message from the slowest consumer's backlog"
)
private BacklogQuota.RetentionPolicy backlogQuotaDefaultRetentionPolicy = BacklogQuota.RetentionPolicy.producer_request_hold;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "Default ttl for namespaces if ttl is not already configured at namespace policies. "
+ "(disable default-ttl with value 0)"
)
private int ttlDurationDefaultInSeconds = 0;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "Enable the deletion of inactive topics.\n"
+ "If only enable this option, will not clean the metadata of partitioned topic."
)
private boolean brokerDeleteInactiveTopicsEnabled = true;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "Metadata of inactive partitioned topic will not be automatically cleaned up by default.\n"
+ "Note: If `allowAutoTopicCreation` and this option are enabled at the same time,\n"
+ "it may appear that a partitioned topic has just been deleted but is automatically created as a non-partitioned topic."
)
private boolean brokerDeleteInactivePartitionedTopicMetadataEnabled = false;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "How often to check for inactive topics"
)
private int brokerDeleteInactiveTopicsFrequencySeconds = 60;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "Set the inactive topic delete mode. Default is delete_when_no_subscriptions\n"
+ "'delete_when_no_subscriptions' mode only delete the topic which has no subscriptions and no active producers\n"
+ "'delete_when_subscriptions_caught_up' mode only delete the topic that all subscriptions has no backlogs(caught up)"
+ "and no active producers/consumers"
)
private InactiveTopicDeleteMode brokerDeleteInactiveTopicsMode = InactiveTopicDeleteMode.delete_when_no_subscriptions;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "Max duration of topic inactivity in seconds, default is not present\n"
+ "If not present, 'brokerDeleteInactiveTopicsFrequencySeconds' will be used\n"
+ "Topics that are inactive for longer than this value will be deleted"
)
private Integer brokerDeleteInactiveTopicsMaxInactiveDurationSeconds = null;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "Allow forced deletion of tenants. Default is false."
)
private boolean forceDeleteTenantAllowed = false;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "Allow forced deletion of namespaces. Default is false."
)
private boolean forceDeleteNamespaceAllowed = false;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "Max pending publish requests per connection to avoid keeping large number of pending "
+ "requests in memory. Default: 1000"
)
private int maxPendingPublishRequestsPerConnection = 1000;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "How frequently to proactively check and purge expired messages"
)
private int messageExpiryCheckIntervalInMinutes = 5;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "How long to delay rewinding cursor and dispatching messages when active consumer is changed"
)
private int activeConsumerFailoverDelayTimeMillis = 1000;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "How long to delete inactive subscriptions from last consuming."
+ " When it is 0, inactive subscriptions are not deleted automatically"
)
private int subscriptionExpirationTimeMinutes = 0;
@FieldContext(
category = CATEGORY_POLICIES,
dynamic = true,
doc = "Enable subscription message redelivery tracker to send redelivery "
+ "count to consumer (default is enabled)"
)
private boolean subscriptionRedeliveryTrackerEnabled = true;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "How frequently to proactively check and purge expired subscription"
)
private int subscriptionExpiryCheckIntervalInMinutes = 5;
@FieldContext(
category = CATEGORY_POLICIES,
dynamic = true,
doc = "Enable subscription types (default is all type enabled)"
)
private Set<String> subscriptionTypesEnabled =
Sets.newHashSet("Exclusive", "Shared", "Failover", "Key_Shared");
@FieldContext(
category = CATEGORY_POLICIES,
dynamic = true,
doc = "Enable Key_Shared subscription (default is enabled)"
)
private boolean subscriptionKeySharedEnable = true;
@FieldContext(category = CATEGORY_POLICIES,
doc = "On KeyShared subscriptions, with default AUTO_SPLIT mode, use splitting ranges or " +
"consistent hashing to reassign keys to new consumers")
private boolean subscriptionKeySharedUseConsistentHashing = false;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "On KeyShared subscriptions, number of points in the consistent-hashing ring. "
+ "The higher the number, the more equal the assignment of keys to consumers")
private int subscriptionKeySharedConsistentHashingReplicaPoints = 100;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "Set the default behavior for message deduplication in the broker.\n\n"
+ "This can be overridden per-namespace. If enabled, broker will reject"
+ " messages that were already stored in the topic"
)
private boolean brokerDeduplicationEnabled = false;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "Maximum number of producer information that it's going to be persisted for deduplication purposes"
)
private int brokerDeduplicationMaxNumberOfProducers = 10000;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "How often is the thread pool scheduled to check whether a snapshot needs to be taken.(disable with value 0)"
)
private int brokerDeduplicationSnapshotFrequencyInSeconds = 120;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "If this time interval is exceeded, a snapshot will be taken."
+ "It will run simultaneously with `brokerDeduplicationEntriesInterval`"
)
private Integer brokerDeduplicationSnapshotIntervalSeconds = 120;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "Number of entries after which a dedup info snapshot is taken.\n\n"
+ "A bigger interval will lead to less snapshots being taken though it would"
+ " increase the topic recovery time, when the entries published after the"
+ " snapshot need to be replayed"
)
private int brokerDeduplicationEntriesInterval = 1000;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "Time of inactivity after which the broker will discard the deduplication information"
+ " relative to a disconnected producer. Default is 6 hours.")
private int brokerDeduplicationProducerInactivityTimeoutMinutes = 360;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "When a namespace is created without specifying the number of bundle, this"
+ " value will be used as the default")
private int defaultNumberOfNamespaceBundles = 4;
@FieldContext(
category = CATEGORY_POLICIES,
dynamic = true,
doc = "The maximum number of namespaces that each tenant can create."
+ "This configuration is not precise control, in a concurrent scenario, the threshold will be exceeded")
private int maxNamespacesPerTenant = 0;
@FieldContext(
category = CATEGORY_POLICIES,
dynamic = true,
doc = "Max number of topics allowed to be created in the namespace. "
+ "When the topics reach the max topics of the namespace, the broker should reject "
+ "the new topic request(include topic auto-created by the producer or consumer) until "
+ "the number of connected consumers decrease. "
+ " Using a value of 0, is disabling maxTopicsPerNamespace-limit check."
)
private int maxTopicsPerNamespace = 0;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "The maximum number of connections in the broker. If it exceeds, new connections are rejected."
)
private int brokerMaxConnections = 0;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "The maximum number of connections per IP. If it exceeds, new connections are rejected."
)
private int brokerMaxConnectionsPerIp = 0;
@FieldContext(
category = CATEGORY_SERVER,
dynamic = true,
doc = "Enable check for minimum allowed client library version"
)
private boolean clientLibraryVersionCheckEnabled = false;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Path for the file used to determine the rotation status for the broker"
+ " when responding to service discovery health checks")
private String statusFilePath;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "Max number of unacknowledged messages allowed to receive messages by a consumer on"
+ " a shared subscription.\n\n Broker will stop sending messages to consumer once,"
+ " this limit reaches until consumer starts acknowledging messages back and unack count"
+ " reaches to `maxUnackedMessagesPerConsumer/2`. Using a value of 0, it is disabling "
+ " unackedMessage-limit check and consumer can receive messages without any restriction")
private int maxUnackedMessagesPerConsumer = 50000;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "Max number of unacknowledged messages allowed per shared subscription. \n\n"
+ " Broker will stop dispatching messages to all consumers of the subscription once this "
+ " limit reaches until consumer starts acknowledging messages back and unack count reaches"
+ " to `limit/2`. Using a value of 0, is disabling unackedMessage-limit check and dispatcher"
+ " can dispatch messages without any restriction")
private int maxUnackedMessagesPerSubscription = 4 * 50000;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "Max number of unacknowledged messages allowed per broker. \n\n"
+ " Once this limit reaches, broker will stop dispatching messages to all shared subscription "
+ " which has higher number of unack messages until subscriptions start acknowledging messages "
+ " back and unack count reaches to `limit/2`. Using a value of 0, is disabling unackedMessage-limit"
+ " check and broker doesn't block dispatchers")
private int maxUnackedMessagesPerBroker = 0;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "Once broker reaches maxUnackedMessagesPerBroker limit, it blocks subscriptions which has higher "
+ " unacked messages than this percentage limit and subscription will not receive any new messages "
+ " until that subscription acks back `limit/2` messages")
private double maxUnackedMessagesPerSubscriptionOnBrokerBlocked = 0.16;
@FieldContext(
category = CATEGORY_POLICIES,
dynamic = true,
doc = "Broker periodically checks if subscription is stuck and unblock if flag is enabled. "
+ "(Default is disabled)"
)
private boolean unblockStuckSubscriptionEnabled = false;
@FieldContext(
category = CATEGORY_POLICIES,
dynamic = true,
doc = "Tick time to schedule task that checks topic publish rate limiting across all topics "
+ "Reducing to lower value can give more accuracy while throttling publish but "
+ "it uses more CPU to perform frequent check. (Disable publish throttling with value 0)"
)
private int topicPublisherThrottlingTickTimeMillis = 5;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Enable precise rate limit for topic publish"
)
private boolean preciseTopicPublishRateLimiterEnable = false;
@FieldContext(
category = CATEGORY_SERVER,
dynamic = true,
doc = "Tick time to schedule task that checks broker publish rate limiting across all topics "
+ "Reducing to lower value can give more accuracy while throttling publish but "
+ "it uses more CPU to perform frequent check. (Disable publish throttling with value 0)"
)
private int brokerPublisherThrottlingTickTimeMillis = 50;
@FieldContext(
category = CATEGORY_SERVER,
dynamic = true,
doc = "Max Rate(in 1 seconds) of Message allowed to publish for a broker "
+ "when broker publish rate limiting enabled. (Disable message rate limit with value 0)"
)
private int brokerPublisherThrottlingMaxMessageRate = 0;
@FieldContext(
category = CATEGORY_SERVER,
dynamic = true,
doc = "Max Rate(in 1 seconds) of Byte allowed to publish for a broker "
+ "when broker publish rate limiting enabled. (Disable byte rate limit with value 0)"
)
private long brokerPublisherThrottlingMaxByteRate = 0;
@FieldContext(
category = CATEGORY_SERVER,
dynamic = true,
doc = "Max Rate(in 1 seconds) of Message allowed to publish for a topic "
+ "when topic publish rate limiting enabled. (Disable byte rate limit with value 0)"
)
private int maxPublishRatePerTopicInMessages = 0;
@FieldContext(
category = CATEGORY_SERVER,
dynamic = true,
doc = "Max Rate(in 1 seconds) of Byte allowed to publish for a topic "
+ "when topic publish rate limiting enabled. (Disable byte rate limit with value 0)"
)
private long maxPublishRatePerTopicInBytes = 0;
@FieldContext(
category = CATEGORY_POLICIES,
dynamic = true,
doc = "Too many subscribe requests from a consumer can cause broker rewinding consumer cursors "
+ " and loading data from bookies, hence causing high network bandwidth usage When the positive"
+ " value is set, broker will throttle the subscribe requests for one consumer. Otherwise, the"
+ " throttling will be disabled. The default value of this setting is 0 - throttling is disabled.")
private int subscribeThrottlingRatePerConsumer = 0;
@FieldContext(
minValue = 1,
dynamic = true,
category = CATEGORY_POLICIES,
doc = "Rate period for {subscribeThrottlingRatePerConsumer}. Default is 30s."
)
private int subscribeRatePeriodPerConsumerInSecond = 30;
@FieldContext(
dynamic = true,
category = CATEGORY_POLICIES,
doc = "Default number of message dispatching throttling-limit for every topic. \n\n"
+ "Using a value of 0, is disabling default message dispatch-throttling")
private int dispatchThrottlingRatePerTopicInMsg = 0;
@FieldContext(
dynamic = true,
category = CATEGORY_POLICIES,
doc = "Default number of message-bytes dispatching throttling-limit for every topic. \n\n"
+ "Using a value of 0, is disabling default message-byte dispatch-throttling")
private long dispatchThrottlingRatePerTopicInByte = 0;
@FieldContext(
dynamic = true,
category = CATEGORY_POLICIES,
doc = "Default number of message dispatching throttling-limit for a subscription. \n\n"
+ "Using a value of 0, is disabling default message dispatch-throttling.")
private int dispatchThrottlingRatePerSubscriptionInMsg = 0;
@FieldContext(
dynamic = true,
category = CATEGORY_POLICIES,
doc = "Default number of message-bytes dispatching throttling-limit for a subscription. \n\n"
+ "Using a value of 0, is disabling default message-byte dispatch-throttling.")
private long dispatchThrottlingRatePerSubscriptionInByte = 0;
@FieldContext(
dynamic = true,
category = CATEGORY_POLICIES,
doc = "Default number of message dispatching throttling-limit for every replicator in replication. \n\n"
+ "Using a value of 0, is disabling replication message dispatch-throttling")
private int dispatchThrottlingRatePerReplicatorInMsg = 0;
@FieldContext(
dynamic = true,
category = CATEGORY_POLICIES,
doc = "Default number of message-bytes dispatching throttling-limit for every replicator in replication. \n\n"
+ "Using a value of 0, is disabling replication message-byte dispatch-throttling")
private long dispatchThrottlingRatePerReplicatorInByte = 0;
@FieldContext(
dynamic = true,
category = CATEGORY_POLICIES,
doc = "Dispatch rate-limiting relative to publish rate. (Enabling flag will make broker to dynamically "
+ "update dispatch-rate relatively to publish-rate: "
+ "throttle-dispatch-rate = (publish-rate + configured dispatch-rate) ")
private boolean dispatchThrottlingRateRelativeToPublishRate = false;
@FieldContext(
dynamic = true,
category = CATEGORY_POLICIES,
doc = "Default dispatch-throttling is disabled for consumers which already caught-up with"
+ " published messages and don't have backlog. This enables dispatch-throttling for "
+ " non-backlog consumers as well.")
private boolean dispatchThrottlingOnNonBacklogConsumerEnabled = false;
@FieldContext(
category = CATEGORY_POLICIES,
doc = "Default policy for publishing usage reports to system topic is disabled."
+ "This enables publishing of usage reports"
)
private String resourceUsageTransportClassName = "";
@FieldContext(
dynamic = true,
category = CATEGORY_POLICIES,
doc = "Default interval to publish usage reports if resourceUsagePublishToTopic is enabled."
)
private int resourceUsageTransportPublishIntervalInSecs = 60;
// <-- dispatcher read settings -->
@FieldContext(
dynamic = true,
category = CATEGORY_SERVER,
doc = "Max number of entries to read from bookkeeper. By default it is 100 entries."
)
private int dispatcherMaxReadBatchSize = 100;
// <-- dispatcher read settings -->
@FieldContext(
dynamic = true,
category = CATEGORY_SERVER,
doc = "Max size in bytes of entries to read from bookkeeper. By default it is 5MB."
)
private int dispatcherMaxReadSizeBytes = 5 * 1024 * 1024;
@FieldContext(
dynamic = true,
category = CATEGORY_SERVER,
doc = "Min number of entries to read from bookkeeper. By default it is 1 entries."
+ "When there is an error occurred on reading entries from bookkeeper, the broker"
+ " will backoff the batch size to this minimum number."
)
private int dispatcherMinReadBatchSize = 1;
@FieldContext(
dynamic = true,
category = CATEGORY_SERVER,
doc = "Max number of entries to dispatch for a shared subscription. By default it is 20 entries."
)
private int dispatcherMaxRoundRobinBatchSize = 20;
@FieldContext(
dynamic = true,
category = CATEGORY_SERVER,
doc = "Precise dispatcher flow control according to history message number of each entry"
)
private boolean preciseDispatcherFlowControl = false;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Whether to use streaming read dispatcher. Currently is in preview and can be changed " +
"in subsequent release."
)
private boolean streamingDispatch = false;
@FieldContext(
dynamic = true,
category = CATEGORY_SERVER,
doc = "Max number of concurrent lookup request broker allows to throttle heavy incoming lookup traffic")
private int maxConcurrentLookupRequest = 50000;
@FieldContext(
dynamic = true,
category = CATEGORY_SERVER,
doc = "Max number of concurrent topic loading request broker allows to control number of zk-operations"
)
private int maxConcurrentTopicLoadRequest = 5000;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Max concurrent non-persistent message can be processed per connection")
private int maxConcurrentNonPersistentMessagePerConnection = 1000;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Number of worker threads to serve non-persistent topic")
private int numWorkerThreadsForNonPersistentTopic = Runtime.getRuntime().availableProcessors();
@FieldContext(
category = CATEGORY_SERVER,
doc = "Enable broker to load persistent topics"
)
private boolean enablePersistentTopics = true;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Enable broker to load non-persistent topics"
)
private boolean enableNonPersistentTopics = true;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Enable to run bookie along with broker"
)
private boolean enableRunBookieTogether = false;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Enable to run bookie autorecovery along with broker"
)
private boolean enableRunBookieAutoRecoveryTogether = false;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Max number of producers allowed to connect to topic. \n\nOnce this limit reaches,"
+ " Broker will reject new producers until the number of connected producers decrease."
+ " Using a value of 0, is disabling maxProducersPerTopic-limit check.")
private int maxProducersPerTopic = 0;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Max number of producers with the same IP address allowed to connect to topic."
+ " \n\nOnce this limit reaches, Broker will reject new producers until the number of"
+ " connected producers with the same IP address decrease."
+ " Using a value of 0, is disabling maxSameAddressProducersPerTopic-limit check.")
private int maxSameAddressProducersPerTopic = 0;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Enforce producer to publish encrypted messages.(default disable).")
private boolean encryptionRequireOnProducer = false;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Max number of consumers allowed to connect to topic. \n\nOnce this limit reaches,"
+ " Broker will reject new consumers until the number of connected consumers decrease."
+ " Using a value of 0, is disabling maxConsumersPerTopic-limit check.")
private int maxConsumersPerTopic = 0;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Max number of consumers with the same IP address allowed to connect to topic."
+ " \n\nOnce this limit reaches, Broker will reject new consumers until the number of"
+ " connected consumers with the same IP address decrease."
+ " Using a value of 0, is disabling maxSameAddressConsumersPerTopic-limit check.")
private int maxSameAddressConsumersPerTopic = 0;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Max number of subscriptions allowed to subscribe to topic. \n\nOnce this limit reaches, "
+ " broker will reject new subscription until the number of subscribed subscriptions decrease.\n"
+ " Using a value of 0, is disabling maxSubscriptionsPerTopic limit check."
)
private int maxSubscriptionsPerTopic = 0;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Max number of consumers allowed to connect to subscription. \n\nOnce this limit reaches,"
+ " Broker will reject new consumers until the number of connected consumers decrease."
+ " Using a value of 0, is disabling maxConsumersPerSubscription-limit check.")
private int maxConsumersPerSubscription = 0;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Max size of messages.",
maxValue = Integer.MAX_VALUE - Commands.MESSAGE_SIZE_FRAME_PADDING)
private int maxMessageSize = Commands.DEFAULT_MAX_MESSAGE_SIZE;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Enable tracking of replicated subscriptions state across clusters.")
private boolean enableReplicatedSubscriptions = true;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Frequency of snapshots for replicated subscriptions tracking.")
private int replicatedSubscriptionsSnapshotFrequencyMillis = 1_000;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Timeout for building a consistent snapshot for tracking replicated subscriptions state. ")
private int replicatedSubscriptionsSnapshotTimeoutSeconds = 30;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Max number of snapshot to be cached per subscription.")
private int replicatedSubscriptionsSnapshotMaxCachedPerSubscription = 10;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Max memory size for broker handling messages sending from producers.\n\n"
+ " If the processing message size exceed this value, broker will stop read data"
+ " from the connection. The processing messages means messages are sends to broker"
+ " but broker have not send response to client, usually waiting to write to bookies.\n\n"
+ " It's shared across all the topics running in the same broker.\n\n"
+ " Use -1 to disable the memory limitation. Default is 1/2 of direct memory.\n\n")
private int maxMessagePublishBufferSizeInMB = Math.max(64,
(int) (PlatformDependent.maxDirectMemory() / 2 / (1024 * 1024)));
@FieldContext(
category = CATEGORY_SERVER,
doc = "Interval between checks to see if message publish buffer size is exceed the max message publish buffer size"
)
private int messagePublishBufferCheckIntervalInMillis = 100;
@FieldContext(category = CATEGORY_SERVER, doc = "Whether to recover cursors lazily when trying to recover a " +
"managed ledger backing a persistent topic. It can improve write availability of topics.\n" +
"The caveat is now when recovered ledger is ready to write we're not sure if all old consumers last mark " +
"delete position can be recovered or not.")
private boolean lazyCursorRecovery = false;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Check between intervals to see if consumed ledgers need to be trimmed"
)
private int retentionCheckIntervalInSeconds = 120;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Check between intervals to see if max message size of topic policy has updated. default is 60s"
)
private int maxMessageSizeCheckIntervalInSeconds = 60;
@FieldContext(
category = CATEGORY_SERVER,
doc = "The number of partitions per partitioned topic.\n"
+ "If try to create or update partitioned topics by exceeded number of partitions, then fail."
)
private int maxNumPartitionsPerPartitionedTopic = 0;
@FieldContext(
category = CATEGORY_SERVER,
doc = "The directory to locate broker interceptors"
)
private String brokerInterceptorsDirectory = "./interceptors";
@FieldContext(
category = CATEGORY_SERVER,
doc = "List of broker interceptor to load, which is a list of broker interceptor names"
)
private Set<String> brokerInterceptors = Sets.newTreeSet();
@FieldContext(
category = CATEGORY_SERVER,
doc = "Enable or disable the broker interceptor, which is only used for testing for now"
)
private boolean disableBrokerInterceptors = true;
@FieldContext(
doc = "There are two policies when zookeeper session expired happens, \"shutdown\" and \"reconnect\". \n\n"
+ " If uses \"shutdown\" policy, shutdown the broker when zookeeper session expired happens.\n\n"
+ " If uses \"reconnect\" policy, try to reconnect to zookeeper server and re-register metadata to zookeeper."
)
private String zookeeperSessionExpiredPolicy = "shutdown";
@FieldContext(
category = CATEGORY_SERVER,
doc = "If a topic remains fenced for this number of seconds, it will be closed forcefully.\n"
+ " If it is set to 0 or a negative number, the fenced topic will not be closed."
)
private int topicFencingTimeoutSeconds = 0;
/**** --- Messaging Protocols --- ****/
@FieldContext(
category = CATEGORY_PROTOCOLS,
doc = "The directory to locate messaging protocol handlers"
)
private String protocolHandlerDirectory = "./protocols";
@FieldContext(
category = CATEGORY_PROTOCOLS,
doc = "List of messaging protocols to load, which is a list of protocol names"
)
private Set<String> messagingProtocols = Sets.newTreeSet();
@FieldContext(
category = CATEGORY_SERVER,
doc = "Enable or disable system topic.")
private boolean systemTopicEnabled = false;
@FieldContext(
category = CATEGORY_SERVER,
doc = "Enable or disable topic level policies, topic level policies depends on the system topic, " +
"please enable the system topic first.")
private boolean topicLevelPoliciesEnabled = false;