Page History

MultiClusterKafkaSource.<String>builder()
  // some default implementations will be provided (file based, statically defined streams)
  .setKafkaMetadataService(new KafkaMetadataServiceImpl())
  .setStreamIds(List.of("my-stream-1", "my-stream-2"))
  .setGroupId("myConsumerGroup")
  .setDeserializer(KafkaRecordDeserializationSchema.valueOnly(StringDeserializer.class))
  .setStartingOffsets(OffsetsInitializer.earliest())
  .setProperties(properties)
  .build();

// Can pass via properties
ConfigOptions.key("multi.cluster.kafka.source.kafka-metadata-service-discovery-internal-ms")
	.longType()
	.noDefaultValue()
	.withDescription(
		"The rate at which the Kafka metadata service will be polled in milliseconds.");