1 package backtype.storm;
2
3 import backtype.storm.ConfigValidation;
4 import backtype.storm.serialization.IKryoDecorator;
5 import backtype.storm.serialization.IKryoFactory;
6 import com.esotericsoftware.kryo.Serializer;
7 import java.util.ArrayList;
8 import java.util.HashMap;
9 import java.util.List;
10 import java.util.Map;
11
12 /**
13 * Topology configs are specified as a plain old map. This class provides a
14 * convenient way to create a topology config map by providing setter methods for
15 * all the configs that can be set. It also makes it easier to do things like add
16 * serializations.
17 *
18 * <p>This class also provides constants for all the configurations possible on
19 * a Storm cluster and Storm topology. Each constant is paired with a schema
20 * that defines the validity criterion of the corresponding field. Default
21 * values for these configs can be found in defaults.yaml.</p>
22 *
23 * <p>Note that you may put other configurations in any of the configs. Storm
24 * will ignore anything it doesn't recognize, but your topologies are free to make
25 * use of them by reading them in the prepare method of Bolts or the open method of
26 * Spouts.</p>
27 */
28 public class Config extends HashMap<String, Object> {
29 /**
30 * The transporter for communication among Storm tasks
31 */
32 public static final String STORM_MESSAGING_TRANSPORT = "storm.messaging.transport";
33 public static final Object STORM_MESSAGING_TRANSPORT_SCHEMA = String.class;
34
35 /**
36 * Netty based messaging: The buffer size for send/recv buffer
37 */
38 public static final String STORM_MESSAGING_NETTY_BUFFER_SIZE = "storm.messaging.netty.buffer_size";
39 public static final Object STORM_MESSAGING_NETTY_BUFFER_SIZE_SCHEMA = Number.class;
40
41 /**
42 * Netty based messaging: The max # of retries that a peer will perform when a remote is not accessible
43 */
44 public static final String STORM_MESSAGING_NETTY_MAX_RETRIES = "storm.messaging.netty.max_retries";
45 public static final Object STORM_MESSAGING_NETTY_MAX_RETRIES_SCHEMA = Number.class;
46
47 /**
48 * Netty based messaging: The min # of milliseconds that a peer will wait.
49 */
50 public static final String STORM_MESSAGING_NETTY_MIN_SLEEP_MS = "storm.messaging.netty.min_wait_ms";
51 public static final Object STORM_MESSAGING_NETTY_MIN_SLEEP_MS_SCHEMA = Number.class;
52
53 /**
54 * Netty based messaging: The max # of milliseconds that a peer will wait.
55 */
56 public static final String STORM_MESSAGING_NETTY_MAX_SLEEP_MS = "storm.messaging.netty.max_wait_ms";
57 public static final Object STORM_MESSAGING_NETTY_MAX_SLEEP_MS_SCHEMA = Number.class;
58
59 /**
60 * Netty based messaging: The # of worker threads for the server.
61 */
62 public static final String STORM_MESSAGING_NETTY_SERVER_WORKER_THREADS = "storm.messaging.netty.server_worker_threads";
63 public static final Object STORM_MESSAGING_NETTY_SERVER_WORKER_THREADS_SCHEMA = Number.class;
64
65 /**
66 * Netty based messaging: The # of worker threads for the client.
67 */
68 public static final String STORM_MESSAGING_NETTY_CLIENT_WORKER_THREADS = "storm.messaging.netty.client_worker_threads";
69 public static final Object STORM_MESSAGING_NETTY_CLIENT_WORKER_THREADS_SCHEMA = Number.class;
70
71 /**
72 * A list of hosts of ZooKeeper servers used to manage the cluster.
73 */
74 public static final String STORM_ZOOKEEPER_SERVERS = "storm.zookeeper.servers";
75 public static final Object STORM_ZOOKEEPER_SERVERS_SCHEMA = ConfigValidation.StringsValidator;
76
77 /**
78 * The port Storm will use to connect to each of the ZooKeeper servers.
79 */
80 public static final String STORM_ZOOKEEPER_PORT = "storm.zookeeper.port";
81 public static final Object STORM_ZOOKEEPER_PORT_SCHEMA = Number.class;
82
83 /**
84 * A directory on the local filesystem used by Storm for any local
85 * filesystem usage it needs. The directory must exist and the Storm daemons must
86 * have permission to read/write from this location.
87 */
88 public static final String STORM_LOCAL_DIR = "storm.local.dir";
89 public static final Object STORM_LOCAL_DIR_SCHEMA = String.class;
90
91 /**
92 * A global task scheduler used to assign topologies's tasks to supervisors' wokers.
93 *
94 * If this is not set, a default system scheduler will be used.
95 */
96 public static final String STORM_SCHEDULER = "storm.scheduler";
97 public static final Object STORM_SCHEDULER_SCHEMA = String.class;
98
99 /**
100 * The mode this Storm cluster is running in. Either "distributed" or "local".
101 */
102 public static final String STORM_CLUSTER_MODE = "storm.cluster.mode";
103 public static final Object STORM_CLUSTER_MODE_SCHEMA = String.class;
104
105 /**
106 * The hostname the supervisors/workers should report to nimbus. If unset, Storm will
107 * get the hostname to report by calling <code>InetAddress.getLocalHost().getCanonicalHostName()</code>.
108 *
109 * You should set this config when you dont have a DNS which supervisors/workers
110 * can utilize to find each other based on hostname got from calls to
111 * <code>InetAddress.getLocalHost().getCanonicalHostName()</code>.
112 */
113 public static final String STORM_LOCAL_HOSTNAME = "storm.local.hostname";
114 public static final Object STORM_LOCAL_HOSTNAME_SCHEMA = String.class;
115
116 /**
117 * The transport plug-in for Thrift client/server communication
118 */
119 public static final String STORM_THRIFT_TRANSPORT_PLUGIN = "storm.thrift.transport";
120 public static final Object STORM_THRIFT_TRANSPORT_PLUGIN_SCHEMA = String.class;
121
122 /**
123 * The serializer class for ListDelegate (tuple payload).
124 * The default serializer will be ListDelegateSerializer
125 */
126 public static final String TOPOLOGY_TUPLE_SERIALIZER = "topology.tuple.serializer";
127 public static final Object TOPOLOGY_TUPLE_SERIALIZER_SCHEMA = String.class;
128
129 /**
130 * Whether or not to use ZeroMQ for messaging in local mode. If this is set
131 * to false, then Storm will use a pure-Java messaging system. The purpose
132 * of this flag is to make it easy to run Storm in local mode by eliminating
133 * the need for native dependencies, which can be difficult to install.
134 *
135 * Defaults to false.
136 */
137 public static final String STORM_LOCAL_MODE_ZMQ = "storm.local.mode.zmq";
138 public static final Object STORM_LOCAL_MODE_ZMQ_SCHEMA = Boolean.class;
139
140 /**
141 * The root location at which Storm stores data in ZooKeeper.
142 */
143 public static final String STORM_ZOOKEEPER_ROOT = "storm.zookeeper.root";
144 public static final Object STORM_ZOOKEEPER_ROOT_SCHEMA = String.class;
145
146 /**
147 * The session timeout for clients to ZooKeeper.
148 */
149 public static final String STORM_ZOOKEEPER_SESSION_TIMEOUT = "storm.zookeeper.session.timeout";
150 public static final Object STORM_ZOOKEEPER_SESSION_TIMEOUT_SCHEMA = Number.class;
151
152 /**
153 * The connection timeout for clients to ZooKeeper.
154 */
155 public static final String STORM_ZOOKEEPER_CONNECTION_TIMEOUT = "storm.zookeeper.connection.timeout";
156 public static final Object STORM_ZOOKEEPER_CONNECTION_TIMEOUT_SCHEMA = Number.class;
157
158 /**
159 * The number of times to retry a Zookeeper operation.
160 */
161 public static final String STORM_ZOOKEEPER_RETRY_TIMES="storm.zookeeper.retry.times";
162 public static final Object STORM_ZOOKEEPER_RETRY_TIMES_SCHEMA = Number.class;
163
164 /**
165 * The interval between retries of a Zookeeper operation.
166 */
167 public static final String STORM_ZOOKEEPER_RETRY_INTERVAL="storm.zookeeper.retry.interval";
168 public static final Object STORM_ZOOKEEPER_RETRY_INTERVAL_SCHEMA = Number.class;
169
170 /**
171 * The ceiling of the interval between retries of a Zookeeper operation.
172 */
173 public static final String STORM_ZOOKEEPER_RETRY_INTERVAL_CEILING="storm.zookeeper.retry.intervalceiling.millis";
174 public static final Object STORM_ZOOKEEPER_RETRY_INTERVAL_CEILING_SCHEMA = Number.class;
175
176 /**
177 * The Zookeeper authentication scheme to use, e.g. "digest". Defaults to no authentication.
178 */
179 public static final String STORM_ZOOKEEPER_AUTH_SCHEME="storm.zookeeper.auth.scheme";
180 public static final Object STORM_ZOOKEEPER_AUTH_SCHEME_SCHEMA = String.class;
181
182 /**
183 * A string representing the payload for Zookeeper authentication. It gets serialized using UTF-8 encoding during authentication.
184 */
185 public static final String STORM_ZOOKEEPER_AUTH_PAYLOAD="storm.zookeeper.auth.payload";
186 public static final Object STORM_ZOOKEEPER_AUTH_PAYLOAD_SCHEMA = String.class;
187
188 /**
189 * The id assigned to a running topology. The id is the storm name with a unique nonce appended.
190 */
191 public static final String STORM_ID = "storm.id";
192 public static final Object STORM_ID_SCHEMA = String.class;
193
194 /**
195 * The host that the master server is running on.
196 */
197 public static final String NIMBUS_HOST = "nimbus.host";
198 public static final Object NIMBUS_HOST_SCHEMA = String.class;
199
200 /**
201 * Which port the Thrift interface of Nimbus should run on. Clients should
202 * connect to this port to upload jars and submit topologies.
203 */
204 public static final String NIMBUS_THRIFT_PORT = "nimbus.thrift.port";
205 public static final Object NIMBUS_THRIFT_PORT_SCHEMA = Number.class;
206
207
208 /**
209 * This parameter is used by the storm-deploy project to configure the
210 * jvm options for the nimbus daemon.
211 */
212 public static final String NIMBUS_CHILDOPTS = "nimbus.childopts";
213 public static final Object NIMBUS_CHILDOPTS_SCHEMA = String.class;
214
215
216 /**
217 * How long without heartbeating a task can go before nimbus will consider the
218 * task dead and reassign it to another location.
219 */
220 public static final String NIMBUS_TASK_TIMEOUT_SECS = "nimbus.task.timeout.secs";
221 public static final Object NIMBUS_TASK_TIMEOUT_SECS_SCHEMA = Number.class;
222
223
224 /**
225 * How often nimbus should wake up to check heartbeats and do reassignments. Note
226 * that if a machine ever goes down Nimbus will immediately wake up and take action.
227 * This parameter is for checking for failures when there's no explicit event like that
228 * occuring.
229 */
230 public static final String NIMBUS_MONITOR_FREQ_SECS = "nimbus.monitor.freq.secs";
231 public static final Object NIMBUS_MONITOR_FREQ_SECS_SCHEMA = Number.class;
232
233 /**
234 * How often nimbus should wake the cleanup thread to clean the inbox.
235 * @see NIMBUS_INBOX_JAR_EXPIRATION_SECS
236 */
237 public static final String NIMBUS_CLEANUP_INBOX_FREQ_SECS = "nimbus.cleanup.inbox.freq.secs";
238 public static final Object NIMBUS_CLEANUP_INBOX_FREQ_SECS_SCHEMA = Number.class;
239
240 /**
241 * The length of time a jar file lives in the inbox before being deleted by the cleanup thread.
242 *
243 * Probably keep this value greater than or equal to NIMBUS_CLEANUP_INBOX_JAR_EXPIRATION_SECS.
244 * Note that the time it takes to delete an inbox jar file is going to be somewhat more than
245 * NIMBUS_CLEANUP_INBOX_JAR_EXPIRATION_SECS (depending on how often NIMBUS_CLEANUP_FREQ_SECS
246 * is set to).
247 * @see NIMBUS_CLEANUP_FREQ_SECS
248 */
249 public static final String NIMBUS_INBOX_JAR_EXPIRATION_SECS = "nimbus.inbox.jar.expiration.secs";
250 public static final Object NIMBUS_INBOX_JAR_EXPIRATION_SECS_SCHEMA = Number.class;
251
252 /**
253 * How long before a supervisor can go without heartbeating before nimbus considers it dead
254 * and stops assigning new work to it.
255 */
256 public static final String NIMBUS_SUPERVISOR_TIMEOUT_SECS = "nimbus.supervisor.timeout.secs";
257 public static final Object NIMBUS_SUPERVISOR_TIMEOUT_SECS_SCHEMA = Number.class;
258
259 /**
260 * A special timeout used when a task is initially launched. During launch, this is the timeout
261 * used until the first heartbeat, overriding nimbus.task.timeout.secs.
262 *
263 * <p>A separate timeout exists for launch because there can be quite a bit of overhead
264 * to launching new JVM's and configuring them.</p>
265 */
266 public static final String NIMBUS_TASK_LAUNCH_SECS = "nimbus.task.launch.secs";
267 public static final Object NIMBUS_TASK_LAUNCH_SECS_SCHEMA = Number.class;
268
269 /**
270 * Whether or not nimbus should reassign tasks if it detects that a task goes down.
271 * Defaults to true, and it's not recommended to change this value.
272 */
273 public static final String NIMBUS_REASSIGN = "nimbus.reassign";
274 public static final Object NIMBUS_REASSIGN_SCHEMA = Boolean.class;
275
276 /**
277 * During upload/download with the master, how long an upload or download connection is idle
278 * before nimbus considers it dead and drops the connection.
279 */
280 public static final String NIMBUS_FILE_COPY_EXPIRATION_SECS = "nimbus.file.copy.expiration.secs";
281 public static final Object NIMBUS_FILE_COPY_EXPIRATION_SECS_SCHEMA = Number.class;
282
283 /**
284 * A custom class that implements ITopologyValidator that is run whenever a
285 * topology is submitted. Can be used to provide business-specific logic for
286 * whether topologies are allowed to run or not.
287 */
288 public static final String NIMBUS_TOPOLOGY_VALIDATOR = "nimbus.topology.validator";
289 public static final Object NIMBUS_TOPOLOGY_VALIDATOR_SCHEMA = String.class;
290
291 /**
292 * Class name for authorization plugin for Nimbus
293 */
294 public static final String NIMBUS_AUTHORIZER = "nimbus.authorizer";
295 public static final Object NIMBUS_AUTHORIZER_SCHEMA = String.class;
296
297 /**
298 * Storm UI binds to this port.
299 */
300 public static final String UI_PORT = "ui.port";
301 public static final Object UI_PORT_SCHEMA = Number.class;
302
303 /**
304 * HTTP UI port for log viewer
305 */
306 public static final String LOGVIEWER_PORT = "logviewer.port";
307 public static final Object LOGVIEWER_PORT_SCHEMA = Number.class;
308
309 /**
310 * Childopts for log viewer java process.
311 */
312 public static final String LOGVIEWER_CHILDOPTS = "logviewer.childopts";
313 public static final Object LOGVIEWER_CHILDOPTS_SCHEMA = String.class;
314
315 /**
316 * Appender name used by log viewer to determine log directory.
317 */
318 public static final String LOGVIEWER_APPENDER_NAME = "logviewer.appender.name";
319 public static final Object LOGVIEWER_APPENDER_NAME_SCHEMA = String.class;
320
321 /**
322 * Childopts for Storm UI Java process.
323 */
324 public static final String UI_CHILDOPTS = "ui.childopts";
325 public static final Object UI_CHILDOPTS_SCHEMA = String.class;
326
327 /**
328 * List of DRPC servers so that the DRPCSpout knows who to talk to.
329 */
330 public static final String DRPC_SERVERS = "drpc.servers";
331 public static final Object DRPC_SERVERS_SCHEMA = ConfigValidation.StringsValidator;
332
333 /**
334 * This port is used by Storm DRPC for receiving DPRC requests from clients.
335 */
336 public static final String DRPC_PORT = "drpc.port";
337 public static final Object DRPC_PORT_SCHEMA = Number.class;
338
339 /**
340 * DRPC thrift server worker threads
341 */
342 public static final String DRPC_WORKER_THREADS = "drpc.worker.threads";
343 public static final Object DRPC_WORKER_THREADS_SCHEMA = Number.class;
344
345 /**
346 * DRPC thrift server queue size
347 */
348 public static final String DRPC_QUEUE_SIZE = "drpc.queue.size";
349 public static final Object DRPC_QUEUE_SIZE_SCHEMA = Number.class;
350
351 /**
352 * This port on Storm DRPC is used by DRPC topologies to receive function invocations and send results back.
353 */
354 public static final String DRPC_INVOCATIONS_PORT = "drpc.invocations.port";
355 public static final Object DRPC_INVOCATIONS_PORT_SCHEMA = Number.class;
356
357 /**
358 * The timeout on DRPC requests within the DRPC server. Defaults to 10 minutes. Note that requests can also
359 * timeout based on the socket timeout on the DRPC client, and separately based on the topology message
360 * timeout for the topology implementing the DRPC function.
361 */
362 public static final String DRPC_REQUEST_TIMEOUT_SECS = "drpc.request.timeout.secs";
363 public static final Object DRPC_REQUEST_TIMEOUT_SECS_SCHEMA = Number.class;
364
365 /**
366 * Childopts for Storm DRPC Java process.
367 */
368 public static final String DRPC_CHILDOPTS = "drpc.childopts";
369 public static final Object DRPC_CHILDOPTS_SCHEMA = String.class;
370
371 /**
372 * the metadata configed on the supervisor
373 */
374 public static final String SUPERVISOR_SCHEDULER_META = "supervisor.scheduler.meta";
375 public static final Object SUPERVISOR_SCHEDULER_META_SCHEMA = Map.class;
376 /**
377 * A list of ports that can run workers on this supervisor. Each worker uses one port, and
378 * the supervisor will only run one worker per port. Use this configuration to tune
379 * how many workers run on each machine.
380 */
381 public static final String SUPERVISOR_SLOTS_PORTS = "supervisor.slots.ports";
382 public static final Object SUPERVISOR_SLOTS_PORTS_SCHEMA = ConfigValidation.NumbersValidator;
383
384
385 /**
386 * This parameter is used by the storm-deploy project to configure the
387 * jvm options for the supervisor daemon.
388 */
389 public static final String SUPERVISOR_CHILDOPTS = "supervisor.childopts";
390 public static final Object SUPERVISOR_CHILDOPTS_SCHEMA = String.class;
391
392
393 /**
394 * How long a worker can go without heartbeating before the supervisor tries to
395 * restart the worker process.
396 */
397 public static final String SUPERVISOR_WORKER_TIMEOUT_SECS = "supervisor.worker.timeout.secs";
398 public static final Object SUPERVISOR_WORKER_TIMEOUT_SECS_SCHEMA = Number.class;
399
400
401 /**
402 * How long a worker can go without heartbeating during the initial launch before
403 * the supervisor tries to restart the worker process. This value override
404 * supervisor.worker.timeout.secs during launch because there is additional
405 * overhead to starting and configuring the JVM on launch.
406 */
407 public static final String SUPERVISOR_WORKER_START_TIMEOUT_SECS = "supervisor.worker.start.timeout.secs";
408 public static final Object SUPERVISOR_WORKER_START_TIMEOUT_SECS_SCHEMA = Number.class;
409
410
411 /**
412 * Whether or not the supervisor should launch workers assigned to it. Defaults
413 * to true -- and you should probably never change this value. This configuration
414 * is used in the Storm unit tests.
415 */
416 public static final String SUPERVISOR_ENABLE = "supervisor.enable";
417 public static final Object SUPERVISOR_ENABLE_SCHEMA = Boolean.class;
418
419
420 /**
421 * how often the supervisor sends a heartbeat to the master.
422 */
423 public static final String SUPERVISOR_HEARTBEAT_FREQUENCY_SECS = "supervisor.heartbeat.frequency.secs";
424 public static final Object SUPERVISOR_HEARTBEAT_FREQUENCY_SECS_SCHEMA = Number.class;
425
426
427 /**
428 * How often the supervisor checks the worker heartbeats to see if any of them
429 * need to be restarted.
430 */
431 public static final String SUPERVISOR_MONITOR_FREQUENCY_SECS = "supervisor.monitor.frequency.secs";
432 public static final Object SUPERVISOR_MONITOR_FREQUENCY_SECS_SCHEMA = Number.class;
433
434 /**
435 * The jvm opts provided to workers launched by this supervisor. All "%ID%" substrings are replaced
436 * with an identifier for this worker.
437 */
438 public static final String WORKER_CHILDOPTS = "worker.childopts";
439 public static final Object WORKER_CHILDOPTS_SCHEMA = String.class;
440
441
442 /**
443 * How often this worker should heartbeat to the supervisor.
444 */
445 public static final String WORKER_HEARTBEAT_FREQUENCY_SECS = "worker.heartbeat.frequency.secs";
446 public static final Object WORKER_HEARTBEAT_FREQUENCY_SECS_SCHEMA = Number.class;
447
448 /**
449 * How often a task should heartbeat its status to the master.
450 */
451 public static final String TASK_HEARTBEAT_FREQUENCY_SECS = "task.heartbeat.frequency.secs";
452 public static final Object TASK_HEARTBEAT_FREQUENCY_SECS_SCHEMA = Number.class;
453
454
455 /**
456 * How often a task should sync its connections with other tasks (if a task is
457 * reassigned, the other tasks sending messages to it need to refresh their connections).
458 * In general though, when a reassignment happens other tasks will be notified
459 * almost immediately. This configuration is here just in case that notification doesn't
460 * come through.
461 */
462 public static final String TASK_REFRESH_POLL_SECS = "task.refresh.poll.secs";
463 public static final Object TASK_REFRESH_POLL_SECS_SCHEMA = Number.class;
464
465
466
467 /**
468 * True if Storm should timeout messages or not. Defaults to true. This is meant to be used
469 * in unit tests to prevent tuples from being accidentally timed out during the test.
470 */
471 public static final String TOPOLOGY_ENABLE_MESSAGE_TIMEOUTS = "topology.enable.message.timeouts";
472 public static final Object TOPOLOGY_ENABLE_MESSAGE_TIMEOUTS_SCHEMA = Boolean.class;
473
474 /**
475 * When set to true, Storm will log every message that's emitted.
476 */
477 public static final String TOPOLOGY_DEBUG = "topology.debug";
478 public static final Object TOPOLOGY_DEBUG_SCHEMA = Boolean.class;
479
480
481 /**
482 * Whether or not the master should optimize topologies by running multiple
483 * tasks in a single thread where appropriate.
484 */
485 public static final String TOPOLOGY_OPTIMIZE = "topology.optimize";
486 public static final Object TOPOLOGY_OPTIMIZE_SCHEMA = Boolean.class;
487
488 /**
489 * How many processes should be spawned around the cluster to execute this
490 * topology. Each process will execute some number of tasks as threads within
491 * them. This parameter should be used in conjunction with the parallelism hints
492 * on each component in the topology to tune the performance of a topology.
493 */
494 public static final String TOPOLOGY_WORKERS = "topology.workers";
495 public static final Object TOPOLOGY_WORKERS_SCHEMA = Number.class;
496
497 /**
498 * How many instances to create for a spout/bolt. A task runs on a thread with zero or more
499 * other tasks for the same spout/bolt. The number of tasks for a spout/bolt is always
500 * the same throughout the lifetime of a topology, but the number of executors (threads) for
501 * a spout/bolt can change over time. This allows a topology to scale to more or less resources
502 * without redeploying the topology or violating the constraints of Storm (such as a fields grouping
503 * guaranteeing that the same value goes to the same task).
504 */
505 public static final String TOPOLOGY_TASKS = "topology.tasks";
506 public static final Object TOPOLOGY_TASKS_SCHEMA = Number.class;
507
508 /**
509 * How many executors to spawn for ackers.
510 *
511 * <p>If this is set to 0, then Storm will immediately ack tuples as soon
512 * as they come off the spout, effectively disabling reliability.</p>
513 */
514 public static final String TOPOLOGY_ACKER_EXECUTORS = "topology.acker.executors";
515 public static final Object TOPOLOGY_ACKER_EXECUTORS_SCHEMA = Number.class;
516
517
518 /**
519 * The maximum amount of time given to the topology to fully process a message
520 * emitted by a spout. If the message is not acked within this time frame, Storm
521 * will fail the message on the spout. Some spouts implementations will then replay
522 * the message at a later time.
523 */
524 public static final String TOPOLOGY_MESSAGE_TIMEOUT_SECS = "topology.message.timeout.secs";
525 public static final Object TOPOLOGY_MESSAGE_TIMEOUT_SECS_SCHEMA = Number.class;
526
527 /**
528 * A list of serialization registrations for Kryo ( http://code.google.com/p/kryo/ ),
529 * the underlying serialization framework for Storm. A serialization can either
530 * be the name of a class (in which case Kryo will automatically create a serializer for the class
531 * that saves all the object's fields), or an implementation of com.esotericsoftware.kryo.Serializer.
532 *
533 * See Kryo's documentation for more information about writing custom serializers.
534 */
535 public static final String TOPOLOGY_KRYO_REGISTER = "topology.kryo.register";
536 public static final Object TOPOLOGY_KRYO_REGISTER_SCHEMA = ConfigValidation.StringsValidator;
537
538 /**
539 * A list of classes that customize storm's kryo instance during start-up.
540 * Each listed class name must implement IKryoDecorator. During start-up the
541 * listed class is instantiated with 0 arguments, then its 'decorate' method
542 * is called with storm's kryo instance as the only argument.
543 */
544 public static final String TOPOLOGY_KRYO_DECORATORS = "topology.kryo.decorators";
545 public static final Object TOPOLOGY_KRYO_DECORATORS_SCHEMA = ConfigValidation.StringsValidator;
546
547 /**
548 * Class that specifies how to create a Kryo instance for serialization. Storm will then apply
549 * topology.kryo.register and topology.kryo.decorators on top of this. The default implementation
550 * implements topology.fall.back.on.java.serialization and turns references off.
551 */
552 public static final String TOPOLOGY_KRYO_FACTORY = "topology.kryo.factory";
553 public static final Object TOPOLOGY_KRYO_FACTORY_SCHEMA = String.class;
554
555
556 /**
557 * Whether or not Storm should skip the loading of kryo registrations for which it
558 * does not know the class or have the serializer implementation. Otherwise, the task will
559 * fail to load and will throw an error at runtime. The use case of this is if you want to
560 * declare your serializations on the storm.yaml files on the cluster rather than every single
561 * time you submit a topology. Different applications may use different serializations and so
562 * a single application may not have the code for the other serializers used by other apps.
563 * By setting this config to true, Storm will ignore that it doesn't have those other serializations
564 * rather than throw an error.
565 */
566 public static final String TOPOLOGY_SKIP_MISSING_KRYO_REGISTRATIONS= "topology.skip.missing.kryo.registrations";
567 public static final Object TOPOLOGY_SKIP_MISSING_KRYO_REGISTRATIONS_SCHEMA = Boolean.class;
568
569 /*
570 * A list of classes implementing IMetricsConsumer (See storm.yaml.example for exact config format).
571 * Each listed class will be routed all the metrics data generated by the storm metrics API.
572 * Each listed class maps 1:1 to a system bolt named __metrics_ClassName#N, and it's parallelism is configurable.
573 */
574 public static final String TOPOLOGY_METRICS_CONSUMER_REGISTER = "topology.metrics.consumer.register";
575 public static final Object TOPOLOGY_METRICS_CONSUMER_REGISTER_SCHEMA = ConfigValidation.MapsValidator;
576
577
578 /**
579 * The maximum parallelism allowed for a component in this topology. This configuration is
580 * typically used in testing to limit the number of threads spawned in local mode.
581 */
582 public static final String TOPOLOGY_MAX_TASK_PARALLELISM="topology.max.task.parallelism";
583 public static final Object TOPOLOGY_MAX_TASK_PARALLELISM_SCHEMA = Number.class;
584
585
586 /**
587 * The maximum number of tuples that can be pending on a spout task at any given time.
588 * This config applies to individual tasks, not to spouts or topologies as a whole.
589 *
590 * A pending tuple is one that has been emitted from a spout but has not been acked or failed yet.
591 * Note that this config parameter has no effect for unreliable spouts that don't tag
592 * their tuples with a message id.
593 */
594 public static final String TOPOLOGY_MAX_SPOUT_PENDING="topology.max.spout.pending";
595 public static final Object TOPOLOGY_MAX_SPOUT_PENDING_SCHEMA = Number.class;
596
597 /**
598 * A class that implements a strategy for what to do when a spout needs to wait. Waiting is
599 * triggered in one of two conditions:
600 *
601 * 1. nextTuple emits no tuples
602 * 2. The spout has hit maxSpoutPending and can't emit any more tuples
603 */
604 public static final String TOPOLOGY_SPOUT_WAIT_STRATEGY="topology.spout.wait.strategy";
605 public static final Object TOPOLOGY_SPOUT_WAIT_STRATEGY_SCHEMA = String.class;
606
607 /**
608 * The amount of milliseconds the SleepEmptyEmitStrategy should sleep for.
609 */
610 public static final String TOPOLOGY_SLEEP_SPOUT_WAIT_STRATEGY_TIME_MS="topology.sleep.spout.wait.strategy.time.ms";
611 public static final Object TOPOLOGY_SLEEP_SPOUT_WAIT_STRATEGY_TIME_MS_SCHEMA = Number.class;
612
613 /**
614 * The maximum amount of time a component gives a source of state to synchronize before it requests
615 * synchronization again.
616 */
617 public static final String TOPOLOGY_STATE_SYNCHRONIZATION_TIMEOUT_SECS="topology.state.synchronization.timeout.secs";
618 public static final Object TOPOLOGY_STATE_SYNCHRONIZATION_TIMEOUT_SECS_SCHEMA = Number.class;
619
620 /**
621 * The percentage of tuples to sample to produce stats for a task.
622 */
623 public static final String TOPOLOGY_STATS_SAMPLE_RATE="topology.stats.sample.rate";
624 public static final Object TOPOLOGY_STATS_SAMPLE_RATE_SCHEMA = Number.class;
625
626 /**
627 * The time period that builtin metrics data in bucketed into.
628 */
629 public static final String TOPOLOGY_BUILTIN_METRICS_BUCKET_SIZE_SECS="topology.builtin.metrics.bucket.size.secs";
630 public static final Object TOPOLOGY_BUILTIN_METRICS_BUCKET_SIZE_SECS_SCHEMA = Number.class;
631
632 /**
633 * Whether or not to use Java serialization in a topology.
634 */
635 public static final String TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION="topology.fall.back.on.java.serialization";
636 public static final Object TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION_SCHEMA = Boolean.class;
637
638 /**
639 * Topology-specific options for the worker child process. This is used in addition to WORKER_CHILDOPTS.
640 */
641 public static final String TOPOLOGY_WORKER_CHILDOPTS="topology.worker.childopts";
642 public static final Object TOPOLOGY_WORKER_CHILDOPTS_SCHEMA = String.class;
643
644 /**
645 * This config is available for TransactionalSpouts, and contains the id ( a String) for
646 * the transactional topology. This id is used to store the state of the transactional
647 * topology in Zookeeper.
648 */
649 public static final String TOPOLOGY_TRANSACTIONAL_ID="topology.transactional.id";
650 public static final Object TOPOLOGY_TRANSACTIONAL_ID_SCHEMA = String.class;
651
652 /**
653 * A list of task hooks that are automatically added to every spout and bolt in the topology. An example
654 * of when you'd do this is to add a hook that integrates with your internal
655 * monitoring system. These hooks are instantiated using the zero-arg constructor.
656 */
657 public static final String TOPOLOGY_AUTO_TASK_HOOKS="topology.auto.task.hooks";
658 public static final Object TOPOLOGY_AUTO_TASK_HOOKS_SCHEMA = ConfigValidation.StringsValidator;
659
660
661 /**
662 * The size of the Disruptor receive queue for each executor. Must be a power of 2.
663 */
664 public static final String TOPOLOGY_EXECUTOR_RECEIVE_BUFFER_SIZE="topology.executor.receive.buffer.size";
665 public static final Object TOPOLOGY_EXECUTOR_RECEIVE_BUFFER_SIZE_SCHEMA = ConfigValidation.PowerOf2Validator;
666
667 /**
668 * The maximum number of messages to batch from the thread receiving off the network to the
669 * executor queues. Must be a power of 2.
670 */
671 public static final String TOPOLOGY_RECEIVER_BUFFER_SIZE="topology.receiver.buffer.size";
672 public static final Object TOPOLOGY_RECEIVER_BUFFER_SIZE_SCHEMA = ConfigValidation.PowerOf2Validator;
673
674 /**
675 * The size of the Disruptor send queue for each executor. Must be a power of 2.
676 */
677 public static final String TOPOLOGY_EXECUTOR_SEND_BUFFER_SIZE="topology.executor.send.buffer.size";
678 public static final Object TOPOLOGY_EXECUTOR_SEND_BUFFER_SIZE_SCHEMA = ConfigValidation.PowerOf2Validator;
679
680 /**
681 * The size of the Disruptor transfer queue for each worker.
682 */
683 public static final String TOPOLOGY_TRANSFER_BUFFER_SIZE="topology.transfer.buffer.size";
684 public static final Object TOPOLOGY_TRANSFER_BUFFER_SIZE_SCHEMA = Number.class;
685
686 /**
687 * How often a tick tuple from the "__system" component and "__tick" stream should be sent
688 * to tasks. Meant to be used as a component-specific configuration.
689 */
690 public static final String TOPOLOGY_TICK_TUPLE_FREQ_SECS="topology.tick.tuple.freq.secs";
691 public static final Object TOPOLOGY_TICK_TUPLE_FREQ_SECS_SCHEMA = Number.class;
692
693
694 /**
695 * Configure the wait strategy used for internal queuing. Can be used to tradeoff latency
696 * vs. throughput
697 */
698 public static final String TOPOLOGY_DISRUPTOR_WAIT_STRATEGY="topology.disruptor.wait.strategy";
699 public static final Object TOPOLOGY_DISRUPTOR_WAIT_STRATEGY_SCHEMA = String.class;
700
701 /**
702 * The size of the shared thread pool for worker tasks to make use of. The thread pool can be accessed
703 * via the TopologyContext.
704 */
705 public static final String TOPOLOGY_WORKER_SHARED_THREAD_POOL_SIZE="topology.worker.shared.thread.pool.size";
706 public static final Object TOPOLOGY_WORKER_SHARED_THREAD_POOL_SIZE_SCHEMA = Number.class;
707
708 /**
709 * The interval in seconds to use for determining whether to throttle error reported to Zookeeper. For example,
710 * an interval of 10 seconds with topology.max.error.report.per.interval set to 5 will only allow 5 errors to be
711 * reported to Zookeeper per task for every 10 second interval of time.
712 */
713 public static final String TOPOLOGY_ERROR_THROTTLE_INTERVAL_SECS="topology.error.throttle.interval.secs";
714 public static final Object TOPOLOGY_ERROR_THROTTLE_INTERVAL_SECS_SCHEMA = Number.class;
715
716 /**
717 * See doc for TOPOLOGY_ERROR_THROTTLE_INTERVAL_SECS
718 */
719 public static final String TOPOLOGY_MAX_ERROR_REPORT_PER_INTERVAL="topology.max.error.report.per.interval";
720 public static final Object TOPOLOGY_MAX_ERROR_REPORT_PER_INTERVAL_SCHEMA = Number.class;
721
722
723 /**
724 * How often a batch can be emitted in a Trident topology.
725 */
726 public static final String TOPOLOGY_TRIDENT_BATCH_EMIT_INTERVAL_MILLIS="topology.trident.batch.emit.interval.millis";
727 public static final Object TOPOLOGY_TRIDENT_BATCH_EMIT_INTERVAL_MILLIS_SCHEMA = Number.class;
728
729 /**
730 * Name of the topology. This config is automatically set by Storm when the topology is submitted.
731 */
732 public static final String TOPOLOGY_NAME="topology.name";
733 public static final Object TOPOLOGY_NAME_SCHEMA = String.class;
734
735 /**
736 * Max pending tuples in one ShellBolt
737 */
738 public static final String TOPOLOGY_SHELLBOLT_MAX_PENDING="topology.shellbolt.max.pending";
739 public static final Object TOPOLOGY_SHELLBOLT_MAX_PENDING_SCHEMA = Number.class;
740
741 /**
742 * The root directory in ZooKeeper for metadata about TransactionalSpouts.
743 */
744 public static final String TRANSACTIONAL_ZOOKEEPER_ROOT="transactional.zookeeper.root";
745 public static final Object TRANSACTIONAL_ZOOKEEPER_ROOT_SCHEMA = String.class;
746
747 /**
748 * The list of zookeeper servers in which to keep the transactional state. If null (which is default),
749 * will use storm.zookeeper.servers
750 */
751 public static final String TRANSACTIONAL_ZOOKEEPER_SERVERS="transactional.zookeeper.servers";
752 public static final Object TRANSACTIONAL_ZOOKEEPER_SERVERS_SCHEMA = ConfigValidation.StringsValidator;
753
754 /**
755 * The port to use to connect to the transactional zookeeper servers. If null (which is default),
756 * will use storm.zookeeper.port
757 */
758 public static final String TRANSACTIONAL_ZOOKEEPER_PORT="transactional.zookeeper.port";
759 public static final Object TRANSACTIONAL_ZOOKEEPER_PORT_SCHEMA = Number.class;
760
761 /**
762 * The number of threads that should be used by the zeromq context in each worker process.
763 */
764 public static final String ZMQ_THREADS = "zmq.threads";
765 public static final Object ZMQ_THREADS_SCHEMA = Number.class;
766
767 /**
768 * How long a connection should retry sending messages to a target host when
769 * the connection is closed. This is an advanced configuration and can almost
770 * certainly be ignored.
771 */
772 public static final String ZMQ_LINGER_MILLIS = "zmq.linger.millis";
773 public static final Object ZMQ_LINGER_MILLIS_SCHEMA = Number.class;
774
775 /**
776 * The high water for the ZeroMQ push sockets used for networking. Use this config to prevent buffer explosion
777 * on the networking layer.
778 */
779 public static final String ZMQ_HWM = "zmq.hwm";
780 public static final Object ZMQ_HWM_SCHEMA = Number.class;
781
782 /**
783 * This value is passed to spawned JVMs (e.g., Nimbus, Supervisor, and Workers)
784 * for the java.library.path value. java.library.path tells the JVM where
785 * to look for native libraries. It is necessary to set this config correctly since
786 * Storm uses the ZeroMQ and JZMQ native libs.
787 */
788 public static final String JAVA_LIBRARY_PATH = "java.library.path";
789 public static final Object JAVA_LIBRARY_PATH_SCHEMA = String.class;
790
791 /**
792 * The path to use as the zookeeper dir when running a zookeeper server via
793 * "storm dev-zookeeper". This zookeeper instance is only intended for development;
794 * it is not a production grade zookeeper setup.
795 */
796 public static final String DEV_ZOOKEEPER_PATH = "dev.zookeeper.path";
797 public static final Object DEV_ZOOKEEPER_PATH_SCHEMA = String.class;
798
799 /**
800 * A map from topology name to the number of machines that should be dedicated for that topology. Set storm.scheduler
801 * to backtype.storm.scheduler.IsolationScheduler to make use of the isolation scheduler.
802 */
803 public static final String ISOLATION_SCHEDULER_MACHINES = "isolation.scheduler.machines";
804 public static final Object ISOLATION_SCHEDULER_MACHINES_SCHEMA = Map.class;
805
806 public static void setDebug(Map conf, boolean isOn) {
807 conf.put(Config.TOPOLOGY_DEBUG, isOn);
808 }
809
810 public void setDebug(boolean isOn) {
811 setDebug(this, isOn);
812 }
813
814 @Deprecated
815 public void setOptimize(boolean isOn) {
816 put(Config.TOPOLOGY_OPTIMIZE, isOn);
817 }
818
819 public static void setNumWorkers(Map conf, int workers) {
820 conf.put(Config.TOPOLOGY_WORKERS, workers);
821 }
822
823 public void setNumWorkers(int workers) {
824 setNumWorkers(this, workers);
825 }
826
827 public static void setNumAckers(Map conf, int numExecutors) {
828 conf.put(Config.TOPOLOGY_ACKER_EXECUTORS, numExecutors);
829 }
830
831 public void setNumAckers(int numExecutors) {
832 setNumAckers(this, numExecutors);
833 }
834
835 public static void setMessageTimeoutSecs(Map conf, int secs) {
836 conf.put(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS, secs);
837 }
838
839 public void setMessageTimeoutSecs(int secs) {
840 setMessageTimeoutSecs(this, secs);
841 }
842
843 public static void registerSerialization(Map conf, Class klass) {
844 getRegisteredSerializations(conf).add(klass.getName());
845 }
846
847 public void registerSerialization(Class klass) {
848 registerSerialization(this, klass);
849 }
850
851 public static void registerSerialization(Map conf, Class klass, Class<? extends Serializer> serializerClass) {
852 Map<String, String> register = new HashMap<String, String>();
853 register.put(klass.getName(), serializerClass.getName());
854 getRegisteredSerializations(conf).add(register);
855 }
856
857 public void registerSerialization(Class klass, Class<? extends Serializer> serializerClass) {
858 registerSerialization(this, klass, serializerClass);
859 }
860
861 public void registerMetricsConsumer(Class klass, Object argument, long parallelismHint) {
862 HashMap m = new HashMap();
863 m.put("class", klass.getCanonicalName());
864 m.put("parallelism.hint", parallelismHint);
865 m.put("argument", argument);
866
867 List l = (List)this.get(TOPOLOGY_METRICS_CONSUMER_REGISTER);
868 if(l == null) { l = new ArrayList(); }
869 l.add(m);
870 this.put(TOPOLOGY_METRICS_CONSUMER_REGISTER, l);
871 }
872
873 public void registerMetricsConsumer(Class klass, long parallelismHint) {
874 registerMetricsConsumer(klass, null, parallelismHint);
875 }
876
877 public void registerMetricsConsumer(Class klass) {
878 registerMetricsConsumer(klass, null, 1L);
879 }
880
881 public static void registerDecorator(Map conf, Class<? extends IKryoDecorator> klass) {
882 getRegisteredDecorators(conf).add(klass.getName());
883 }
884
885 public void registerDecorator(Class<? extends IKryoDecorator> klass) {
886 registerDecorator(this, klass);
887 }
888
889 public static void setKryoFactory(Map conf, Class<? extends IKryoFactory> klass) {
890 conf.put(Config.TOPOLOGY_KRYO_FACTORY, klass.getName());
891 }
892
893 public void setKryoFactory(Class<? extends IKryoFactory> klass) {
894 setKryoFactory(this, klass);
895 }
896
897 public static void setSkipMissingKryoRegistrations(Map conf, boolean skip) {
898 conf.put(Config.TOPOLOGY_SKIP_MISSING_KRYO_REGISTRATIONS, skip);
899 }
900
901 public void setSkipMissingKryoRegistrations(boolean skip) {
902 setSkipMissingKryoRegistrations(this, skip);
903 }
904
905 public static void setMaxTaskParallelism(Map conf, int max) {
906 conf.put(Config.TOPOLOGY_MAX_TASK_PARALLELISM, max);
907 }
908
909 public void setMaxTaskParallelism(int max) {
910 setMaxTaskParallelism(this, max);
911 }
912
913 public static void setMaxSpoutPending(Map conf, int max) {
914 conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, max);
915 }
916
917 public void setMaxSpoutPending(int max) {
918 setMaxSpoutPending(this, max);
919 }
920
921 public static void setStatsSampleRate(Map conf, double rate) {
922 conf.put(Config.TOPOLOGY_STATS_SAMPLE_RATE, rate);
923 }
924
925 public void setStatsSampleRate(double rate) {
926 setStatsSampleRate(this, rate);
927 }
928
929 public static void setFallBackOnJavaSerialization(Map conf, boolean fallback) {
930 conf.put(Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION, fallback);
931 }
932
933 public void setFallBackOnJavaSerialization(boolean fallback) {
934 setFallBackOnJavaSerialization(this, fallback);
935 }
936
937 private static List getRegisteredSerializations(Map conf) {
938 List ret;
939 if(!conf.containsKey(Config.TOPOLOGY_KRYO_REGISTER)) {
940 ret = new ArrayList();
941 } else {
942 ret = new ArrayList((List) conf.get(Config.TOPOLOGY_KRYO_REGISTER));
943 }
944 conf.put(Config.TOPOLOGY_KRYO_REGISTER, ret);
945 return ret;
946 }
947
948 private static List getRegisteredDecorators(Map conf) {
949 List ret;
950 if(!conf.containsKey(Config.TOPOLOGY_KRYO_DECORATORS)) {
951 ret = new ArrayList();
952 } else {
953 ret = new ArrayList((List) conf.get(Config.TOPOLOGY_KRYO_DECORATORS));
954 }
955 conf.put(Config.TOPOLOGY_KRYO_DECORATORS, ret);
956 return ret;
957 }
958 }
|手机版|小黑屋|Java自学者论坛
( 声明：本站文章及资料整理自互联网，用于Java自学者交流学习使用，对资料版权不负任何法律责任，若有侵权请及时联系客服屏蔽删除 )
发表于 2021-8-27 09:48:05
