You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by pm...@apache.org on 2018/08/27 19:06:32 UTC
[2/5] samza git commit: SAMZA-1804: System and Stream Descriptors
http://git-wip-us.apache.org/repos/asf/samza/blob/2a71baf7/samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaInputDescriptor.java
----------------------------------------------------------------------
diff --git a/samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaInputDescriptor.java b/samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaInputDescriptor.java
new file mode 100644
index 0000000..d3fe252
--- /dev/null
+++ b/samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaInputDescriptor.java
@@ -0,0 +1,108 @@
+/*
+ * 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.samza.system.kafka;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.samza.operators.descriptors.base.stream.InputDescriptor;
+import org.apache.samza.operators.descriptors.base.system.SystemDescriptor;
+import org.apache.samza.operators.functions.InputTransformer;
+import org.apache.samza.serializers.Serde;
+
+/**
+ * A descriptor for a kafka input stream.
+ * <p>
+ * An instance of this descriptor may be obtained from an appropriately configured {@link KafkaSystemDescriptor}.
+ * <p>
+ * Stream properties configured using a descriptor override corresponding properties provided in configuration.
+ *
+ * @param <StreamMessageType> type of messages in this stream.
+ */
+public class KafkaInputDescriptor<StreamMessageType>
+ extends InputDescriptor<StreamMessageType, KafkaInputDescriptor<StreamMessageType>> {
+ private static final String CONSUMER_AUTO_OFFSET_RESET_CONFIG_KEY = "systems.%s.streams.%s.consumer.auto.offset.reset";
+ private static final String CONSUMER_FETCH_MESSAGE_MAX_BYTES_CONFIG_KEY = "systems.%s.streams.%s.consumer.fetch.message.max.bytes";
+
+ private Optional<String> consumerAutoOffsetResetOptional = Optional.empty();
+ private Optional<Long> consumerFetchMessageMaxBytesOptional = Optional.empty();
+
+ KafkaInputDescriptor(String streamId, SystemDescriptor systemDescriptor, Serde serde, InputTransformer transformer) {
+ super(streamId, serde, systemDescriptor, transformer);
+ }
+
+ /**
+ * This setting determines what happens if a consumer attempts to read an offset for this topic that is outside of
+ * the current valid range. This could happen if the topic does not exist, or if a checkpoint is older than the
+ * maximum message history retained by the brokers. This property is not to be confused with
+ * {@link InputDescriptor#withOffsetDefault}, which determines what happens if there is no checkpoint.
+ * <p>
+ * The following are valid values for auto.offset.reset:
+ * <ul>
+ * <li>smallest: Start consuming at the smallest (oldest) offset available on the broker
+ * (process as much message history as available).
+ * <li>largest: Start consuming at the largest (newest) offset available on the broker
+ * (skip any messages published while the job was not running).
+ * <li>anything else: Throw an exception and refuse to start up the job.
+ * </ul>
+ * <p>
+ * Note: This property may be set at a system level using {@link KafkaSystemDescriptor#withConsumerAutoOffsetReset}
+ *
+ * @param consumerAutoOffsetReset consumer auto offset reset policy for the input
+ * @return this input descriptor
+ */
+ public KafkaInputDescriptor<StreamMessageType> withConsumerAutoOffsetReset(String consumerAutoOffsetReset) {
+ this.consumerAutoOffsetResetOptional = Optional.of(StringUtils.stripToNull(consumerAutoOffsetReset));
+ return this;
+ }
+
+ /**
+ * The number of bytes of messages to attempt to fetch for each topic-partition for this topic in each fetch request.
+ * These bytes will be read into memory for each partition, so this helps control the memory used by the consumer.
+ * The fetch request size must be at least as large as the maximum message size the server allows or else it is
+ * possible for the producer to send messages larger than the consumer can fetch.
+ * <p>
+ * Note: This property may be set at a system level using {@link KafkaSystemDescriptor#withConsumerFetchMessageMaxBytes}
+ *
+ * @param fetchMessageMaxBytes number of bytes of messages to attempt to fetch for each topic-partition
+ * in each fetch request
+ * @return this input descriptor
+ */
+ public KafkaInputDescriptor<StreamMessageType> withConsumerFetchMessageMaxBytes(long fetchMessageMaxBytes) {
+ this.consumerFetchMessageMaxBytesOptional = Optional.of(fetchMessageMaxBytes);
+ return this;
+ }
+
+ @Override
+ public Map<String, String> toConfig() {
+ HashMap<String, String> configs = new HashMap<>(super.toConfig());
+ // Note: Kafka configuration needs the topic's physical name, not the stream-id.
+ // We won't have that here if user only specified it in configs, or if it got rewritten
+ // by the planner to something different than what's in this descriptor.
+ String streamName = getPhysicalName().orElse(getStreamId());
+ String systemName = getSystemName();
+
+ consumerAutoOffsetResetOptional.ifPresent(autoOffsetReset ->
+ configs.put(String.format(CONSUMER_AUTO_OFFSET_RESET_CONFIG_KEY, systemName, streamName), autoOffsetReset));
+ consumerFetchMessageMaxBytesOptional.ifPresent(fetchMessageMaxBytes ->
+ configs.put(String.format(CONSUMER_FETCH_MESSAGE_MAX_BYTES_CONFIG_KEY, systemName, streamName), Long.toString(fetchMessageMaxBytes)));
+ return configs;
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/2a71baf7/samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaOutputDescriptor.java
----------------------------------------------------------------------
diff --git a/samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaOutputDescriptor.java b/samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaOutputDescriptor.java
new file mode 100644
index 0000000..1142276
--- /dev/null
+++ b/samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaOutputDescriptor.java
@@ -0,0 +1,39 @@
+/*
+ * 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.samza.system.kafka;
+
+import org.apache.samza.operators.descriptors.base.stream.OutputDescriptor;
+import org.apache.samza.operators.descriptors.base.system.SystemDescriptor;
+import org.apache.samza.serializers.Serde;
+
+/**
+ * A descriptor for a kafka output stream.
+ * <p>
+ * An instance of this descriptor may be obtained from an appropriately configured {@link KafkaSystemDescriptor}.
+ * <p>
+ * Stream properties configured using a descriptor override corresponding properties provided in configuration.
+ *
+ * @param <StreamMessageType> type of messages in this stream.
+ */
+public class KafkaOutputDescriptor<StreamMessageType>
+ extends OutputDescriptor<StreamMessageType, KafkaOutputDescriptor<StreamMessageType>> {
+ KafkaOutputDescriptor(String streamId, SystemDescriptor systemDescriptor, Serde<StreamMessageType> serde) {
+ super(streamId, serde, systemDescriptor);
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/2a71baf7/samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemDescriptor.java
----------------------------------------------------------------------
diff --git a/samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemDescriptor.java b/samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemDescriptor.java
new file mode 100644
index 0000000..6fa8915
--- /dev/null
+++ b/samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemDescriptor.java
@@ -0,0 +1,251 @@
+/*
+ * 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.samza.system.kafka;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.samza.config.KafkaConfig;
+import org.apache.samza.operators.descriptors.base.stream.InputDescriptor;
+import org.apache.samza.operators.descriptors.base.system.OutputDescriptorProvider;
+import org.apache.samza.operators.descriptors.base.system.SimpleInputDescriptorProvider;
+import org.apache.samza.operators.descriptors.base.system.SystemDescriptor;
+import org.apache.samza.serializers.Serde;
+
+
+/**
+ * A descriptor for a Kafka system.
+ * <p>
+ * System properties configured using a descriptor override corresponding properties provided in configuration.
+ */
+@SuppressWarnings("unchecked")
+public class KafkaSystemDescriptor extends SystemDescriptor<KafkaSystemDescriptor>
+ implements SimpleInputDescriptorProvider, OutputDescriptorProvider {
+ private static final String FACTORY_CLASS_NAME = KafkaSystemFactory.class.getName();
+ private static final String CONSUMER_ZK_CONNECT_CONFIG_KEY = "systems.%s.consumer.zookeeper.connect";
+ private static final String CONSUMER_AUTO_OFFSET_RESET_CONFIG_KEY = "systems.%s.consumer.auto.offset.reset";
+ private static final String CONSUMER_FETCH_THRESHOLD_CONFIG_KEY = KafkaConfig.CONSUMER_FETCH_THRESHOLD();
+ private static final String CONSUMER_FETCH_THRESHOLD_BYTES_CONFIG_KEY = KafkaConfig.CONSUMER_FETCH_THRESHOLD_BYTES();
+ private static final String CONSUMER_FETCH_MESSAGE_MAX_BYTES_KEY = "systems.%s.consumer.fetch.message.max.bytes";
+ private static final String CONSUMER_CONFIGS_CONFIG_KEY = "systems.%s.consumer.%s";
+ private static final String PRODUCER_BOOTSTRAP_SERVERS_CONFIG_KEY = "systems.%s.producer.bootstrap.servers";
+ private static final String PRODUCER_CONFIGS_CONFIG_KEY = "systems.%s.producer.%s";
+
+ private List<String> consumerZkConnect = Collections.emptyList();
+ private Optional<String> consumerAutoOffsetResetOptional = Optional.empty();
+ private Optional<Integer> consumerFetchThresholdOptional = Optional.empty();
+ private Optional<Long> consumerFetchThresholdBytesOptional = Optional.empty();
+ private Optional<Long> consumerFetchMessageMaxBytesOptional = Optional.empty();
+ private Map<String, String> consumerConfigs = Collections.emptyMap();
+ private List<String> producerBootstrapServers = Collections.emptyList();
+ private Map<String, String> producerConfigs = Collections.emptyMap();
+
+ /**
+ * Constructs a {@link KafkaSystemDescriptor} instance with no system level serde.
+ * Serdes must be provided explicitly at stream level when getting input or output descriptors.
+ *
+ * @param systemName name of this system
+ */
+ public KafkaSystemDescriptor(String systemName) {
+ super(systemName, FACTORY_CLASS_NAME, null, null);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public <StreamMessageType> KafkaInputDescriptor<StreamMessageType> getInputDescriptor(String streamId, Serde<StreamMessageType> serde) {
+ return new KafkaInputDescriptor<>(streamId, this, serde, null);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public <StreamMessageType> KafkaOutputDescriptor<StreamMessageType> getOutputDescriptor(String streamId, Serde<StreamMessageType> serde) {
+ return new KafkaOutputDescriptor<>(streamId, this, serde);
+ }
+
+ /**
+ * The hostname and port of one or more Zookeeper nodes where information about the Kafka cluster can be found.
+ * This is given as a list of hostname:port pairs, such as
+ * {@code ImmutableList.of("zk1.example.com:2181", "zk2.example.com:2181", "zk3.example.com:2181")}.
+ * If the cluster information is at some sub-path of the Zookeeper namespace, you need to include the path at the
+ * end of the list of hostnames, for example:
+ * {@code ImmutableList.of("zk1.example.com:2181", "zk2.example.com:2181/clusters/my-kafka")}.
+ *
+ * @param consumerZkConnect Zookeeper connection information for the system
+ * @return this system descriptor
+ */
+ public KafkaSystemDescriptor withConsumerZkConnect(List<String> consumerZkConnect) {
+ this.consumerZkConnect = consumerZkConnect;
+ return this;
+ }
+
+ /**
+ * This setting determines what happens if a consumer attempts to read an offset that is outside of the current
+ * valid range. This could happen if the topic does not exist, or if a checkpoint is older than the maximum message
+ * history retained by the brokers. This property is not to be confused with {@link InputDescriptor#withOffsetDefault},
+ * which determines what happens if there is no checkpoint.
+ * <p>
+ * The following are valid values for auto.offset.reset:
+ * <ul>
+ * <li>smallest: Start consuming at the smallest (oldest) offset available on the broker
+ * (process as much message history as available).
+ * <li>largest: Start consuming at the largest (newest) offset available on the broker
+ * (skip any messages published while the job was not running).
+ * <li>anything else: Throw an exception and refuse to start up the job.
+ * </ul>
+ * <p>
+ * Note: This property may be set at a topic level using {@link KafkaInputDescriptor#withConsumerAutoOffsetReset}
+ *
+ * @param consumerAutoOffsetReset consumer auto offset reset policy for the system
+ * @return this system descriptor
+ */
+ public KafkaSystemDescriptor withConsumerAutoOffsetReset(String consumerAutoOffsetReset) {
+ this.consumerAutoOffsetResetOptional = Optional.of(StringUtils.stripToNull(consumerAutoOffsetReset));
+ return this;
+ }
+
+ /**
+ * When consuming streams from Kafka, a Samza container maintains an in-memory buffer for incoming messages in
+ * order to increase throughput (the stream task can continue processing buffered messages while new messages are
+ * fetched from Kafka). This parameter determines the number of messages we aim to buffer across all stream partitions
+ * consumed by a container. For example, if a container consumes 50 partitions, it will try to buffer 1000
+ * messages per partition by default. When the number of buffered messages falls below that threshold, Samza
+ * fetches more messages from the Kafka broker to replenish the buffer. Increasing this parameter can increase
+ * a job's processing throughput, but also increases the amount of memory used.
+ *
+ * @param fetchThreshold number of incoming messages to buffer in-memory
+ * @return this system descriptor
+ */
+ public KafkaSystemDescriptor withSamzaFetchThreshold(int fetchThreshold) {
+ this.consumerFetchThresholdOptional = Optional.of(fetchThreshold);
+ return this;
+ }
+
+ /**
+ * When consuming streams from Kafka, a Samza container maintains an in-memory buffer for incoming messages in
+ * order to increase throughput (the stream task can continue processing buffered messages while new messages are
+ * fetched from Kafka). This parameter determines the total size of messages we aim to buffer across all stream
+ * partitions consumed by a container based on bytes. Defines how many bytes to use for the buffered prefetch
+ * messages for job as a whole. The bytes for a single system/stream/partition are computed based on this.
+ * This fetches the entire messages, hence this bytes limit is a soft one, and the actual usage can be
+ * the bytes limit + size of max message in the partition for a given stream. If the value of this property
+ * is > 0 then this takes precedence over systems.system-name.samza.fetch.threshold.
+ * <p>
+ * For example, if fetchThresholdBytes is set to 100000 bytes, and there are 50 SystemStreamPartitions registered,
+ * then the per-partition threshold is (100000 / 2) / 50 = 1000 bytes. As this is a soft limit, the actual usage
+ * can be 1000 bytes + size of max message. As soon as a SystemStreamPartition's buffered messages bytes drops
+ * below 1000, a fetch request will be executed to get more data for it. Increasing this parameter will decrease
+ * the latency between when a queue is drained of messages and when new messages are enqueued, but also leads
+ * to an increase in memory usage since more messages will be held in memory. The default value is -1,
+ * which means this is not used.
+ *
+ * @param fetchThresholdBytes number of bytes for incoming messages to buffer in-memory
+ * @return this system descriptor
+ */
+ public KafkaSystemDescriptor withSamzaFetchThresholdBytes(long fetchThresholdBytes) {
+ this.consumerFetchThresholdBytesOptional = Optional.of(fetchThresholdBytes);
+ return this;
+ }
+
+ /**
+ * The number of bytes of messages to attempt to fetch for each topic-partition in each fetch request.
+ * These bytes will be read into memory for each partition, so this helps control the memory used by the consumer.
+ * The fetch request size must be at least as large as the maximum message size the server allows or else it is
+ * possible for the producer to send messages larger than the consumer can fetch.
+ * <p>
+ * Note: This property may be set at a topic level using {@link KafkaInputDescriptor#withConsumerFetchMessageMaxBytes}
+ *
+ * @param fetchMessageMaxBytes number of bytes of messages to attempt to fetch for each topic-partition
+ * in each fetch request
+ * @return this system descriptor
+ */
+ public KafkaSystemDescriptor withConsumerFetchMessageMaxBytes(long fetchMessageMaxBytes) {
+ this.consumerFetchMessageMaxBytesOptional = Optional.of(fetchMessageMaxBytes);
+ return this;
+ }
+
+ /**
+ * Any Kafka consumer configuration can be included here. For example, to change the socket timeout,
+ * you can set socket.timeout.ms. (There is no need to configure group.id or client.id, as they are automatically
+ * configured by Samza. Also, there is no need to set auto.commit.enable because Samza has its own
+ * checkpointing mechanism.)
+ *
+ * @param consumerConfigs additional consumer configuration
+ * @return this system descriptor
+ */
+ public KafkaSystemDescriptor withConsumerConfigs(Map<String, String> consumerConfigs) {
+ this.consumerConfigs = consumerConfigs;
+ return this;
+ }
+
+ /**
+ * A list of network endpoints where the Kafka brokers are running. This is given as a list of hostname:port pairs,
+ * for example {@code ImmutableList.of("kafka1.example.com:9092", "kafka2.example.com:9092", "kafka3.example.com:9092")}.
+ * It's not necessary to list every single Kafka node in the cluster: Samza uses this property in order to discover
+ * which topics and partitions are hosted on which broker. This property is needed even if you are only consuming
+ * from Kafka, and not writing to it, because Samza uses it to discover metadata about streams being consumed.
+ *
+ * @param producerBootstrapServers network endpoints where the kafka brokers are running
+ * @return this system descriptor
+ */
+ public KafkaSystemDescriptor withProducerBootstrapServers(List<String> producerBootstrapServers) {
+ this.producerBootstrapServers = producerBootstrapServers;
+ return this;
+ }
+
+ /**
+ * Any Kafka producer configuration can be included here. For example, to change the request timeout,
+ * you can set timeout.ms. (There is no need to configure client.id as it is automatically configured by Samza.)
+ *
+ * @param producerConfigs additional producer configuration
+ * @return this system descriptor
+ */
+ public KafkaSystemDescriptor withProducerConfigs(Map<String, String> producerConfigs) {
+ this.producerConfigs = producerConfigs;
+ return this;
+ }
+
+ @Override
+ public Map<String, String> toConfig() {
+ Map<String, String> configs = new HashMap<>(super.toConfig());
+ if(!consumerZkConnect.isEmpty()) {
+ configs.put(String.format(CONSUMER_ZK_CONNECT_CONFIG_KEY, getSystemName()), String.join(",", consumerZkConnect));
+ }
+ consumerAutoOffsetResetOptional.ifPresent(consumerAutoOffsetReset ->
+ configs.put(String.format(CONSUMER_AUTO_OFFSET_RESET_CONFIG_KEY, getSystemName()), consumerAutoOffsetReset));
+ consumerFetchThresholdOptional.ifPresent(consumerFetchThreshold ->
+ configs.put(String.format(CONSUMER_FETCH_THRESHOLD_CONFIG_KEY, getSystemName()), Integer.toString(consumerFetchThreshold)));
+ consumerFetchThresholdBytesOptional.ifPresent(consumerFetchThresholdBytes ->
+ configs.put(String.format(CONSUMER_FETCH_THRESHOLD_BYTES_CONFIG_KEY, getSystemName()), Long.toString(consumerFetchThresholdBytes)));
+ consumerFetchMessageMaxBytesOptional.ifPresent(consumerFetchMessageMaxBytes ->
+ configs.put(String.format(CONSUMER_FETCH_MESSAGE_MAX_BYTES_KEY, getSystemName()), Long.toString(consumerFetchMessageMaxBytes)));
+ consumerConfigs.forEach((key, value) -> configs.put(String.format(CONSUMER_CONFIGS_CONFIG_KEY, getSystemName(), key), value));
+ if (!producerBootstrapServers.isEmpty()) {
+ configs.put(String.format(PRODUCER_BOOTSTRAP_SERVERS_CONFIG_KEY, getSystemName()), String.join(",", producerBootstrapServers));
+ }
+ producerConfigs.forEach((key, value) -> configs.put(String.format(PRODUCER_CONFIGS_CONFIG_KEY, getSystemName(), key), value));
+ return configs;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/samza/blob/2a71baf7/samza-kafka/src/test/java/org/apache/samza/system/kafka/TestKafkaInputDescriptor.java
----------------------------------------------------------------------
diff --git a/samza-kafka/src/test/java/org/apache/samza/system/kafka/TestKafkaInputDescriptor.java b/samza-kafka/src/test/java/org/apache/samza/system/kafka/TestKafkaInputDescriptor.java
new file mode 100644
index 0000000..ac90cf0
--- /dev/null
+++ b/samza-kafka/src/test/java/org/apache/samza/system/kafka/TestKafkaInputDescriptor.java
@@ -0,0 +1,68 @@
+/*
+ * 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.samza.system.kafka;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+
+import java.util.Map;
+import org.apache.samza.SamzaException;
+import org.apache.samza.operators.KV;
+import org.apache.samza.operators.descriptors.GenericSystemDescriptor;
+import org.apache.samza.serializers.IntegerSerde;
+import org.apache.samza.serializers.KVSerde;
+import org.apache.samza.serializers.StringSerde;
+import org.apache.samza.system.SystemStreamMetadata;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class TestKafkaInputDescriptor {
+ @Test
+ public void testISDConfigsWithOverrides() {
+ KafkaSystemDescriptor sd = new KafkaSystemDescriptor("kafka");
+
+ KafkaInputDescriptor<KV<String, Integer>> isd =
+ sd.getInputDescriptor("input-stream", KVSerde.of(new StringSerde(), new IntegerSerde()))
+ .withPhysicalName("physical-name")
+ .withConsumerAutoOffsetReset("largest")
+ .withConsumerFetchMessageMaxBytes(1024 * 1024);
+
+ Map<String, String> generatedConfigs = isd.toConfig();;
+ assertEquals("kafka", generatedConfigs.get("streams.input-stream.samza.system"));
+ assertEquals("physical-name", generatedConfigs.get("streams.input-stream.samza.physical.name"));
+ assertEquals("largest", generatedConfigs.get("systems.kafka.streams.physical-name.consumer.auto.offset.reset"));
+ assertEquals("1048576", generatedConfigs.get("systems.kafka.streams.physical-name.consumer.fetch.message.max.bytes"));
+ }
+
+ @Test
+ public void testISDConfigsWithDefaults() {
+ KafkaSystemDescriptor sd = new KafkaSystemDescriptor("kafka")
+ .withConsumerZkConnect(ImmutableList.of("localhost:123"))
+ .withProducerBootstrapServers(ImmutableList.of("localhost:567", "localhost:890"));
+
+ KafkaInputDescriptor<KV<String, Integer>> isd =
+ sd.getInputDescriptor("input-stream", KVSerde.of(new StringSerde(), new IntegerSerde()));
+
+ Map<String, String> generatedConfigs = isd.toConfig();
+ assertEquals("kafka", generatedConfigs.get("streams.input-stream.samza.system"));
+ assertEquals(1, generatedConfigs.size()); // verify that there are no other configs
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/2a71baf7/samza-kafka/src/test/java/org/apache/samza/system/kafka/TestKafkaSystemDescriptor.java
----------------------------------------------------------------------
diff --git a/samza-kafka/src/test/java/org/apache/samza/system/kafka/TestKafkaSystemDescriptor.java b/samza-kafka/src/test/java/org/apache/samza/system/kafka/TestKafkaSystemDescriptor.java
new file mode 100644
index 0000000..e1ddac0
--- /dev/null
+++ b/samza-kafka/src/test/java/org/apache/samza/system/kafka/TestKafkaSystemDescriptor.java
@@ -0,0 +1,69 @@
+/*
+ * 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.samza.system.kafka;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+
+import java.util.Map;
+import org.apache.samza.system.SystemStreamMetadata;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+public class TestKafkaSystemDescriptor {
+ @Test
+ public void testSDConfigsWithOverrides() {
+ KafkaSystemDescriptor sd =
+ new KafkaSystemDescriptor("kafka")
+ .withConsumerZkConnect(ImmutableList.of("localhost:1234"))
+ .withProducerBootstrapServers(ImmutableList.of("localhost:567", "localhost:890"))
+ .withDefaultStreamOffsetDefault(SystemStreamMetadata.OffsetType.OLDEST)
+ .withConsumerAutoOffsetReset("smallest")
+ .withConsumerFetchMessageMaxBytes(1024*1024)
+ .withSamzaFetchThreshold(10000)
+ .withSamzaFetchThresholdBytes(1024 * 1024)
+ .withConsumerConfigs(ImmutableMap.of("custom-consumer-config-key", "custom-consumer-config-value"))
+ .withProducerConfigs(ImmutableMap.of("custom-producer-config-key", "custom-producer-config-value"))
+ .withDefaultStreamConfigs(ImmutableMap.of("custom-stream-config-key", "custom-stream-config-value"));
+
+ Map<String, String> generatedConfigs = sd.toConfig();
+ assertEquals("org.apache.samza.system.kafka.KafkaSystemFactory", generatedConfigs.get("systems.kafka.samza.factory"));
+ assertEquals("localhost:1234", generatedConfigs.get("systems.kafka.consumer.zookeeper.connect"));
+ assertEquals("localhost:567,localhost:890", generatedConfigs.get("systems.kafka.producer.bootstrap.servers"));
+ assertEquals("smallest", generatedConfigs.get("systems.kafka.consumer.auto.offset.reset"));
+ assertEquals("1048576", generatedConfigs.get("systems.kafka.consumer.fetch.message.max.bytes"));
+ assertEquals("10000", generatedConfigs.get("systems.kafka.samza.fetch.threshold"));
+ assertEquals("1048576", generatedConfigs.get("systems.kafka.samza.fetch.threshold.bytes"));
+ assertEquals("custom-consumer-config-value", generatedConfigs.get("systems.kafka.consumer.custom-consumer-config-key"));
+ assertEquals("custom-producer-config-value", generatedConfigs.get("systems.kafka.producer.custom-producer-config-key"));
+ assertEquals("custom-stream-config-value", generatedConfigs.get("systems.kafka.default.stream.custom-stream-config-key"));
+ assertEquals("oldest", generatedConfigs.get("systems.kafka.default.stream.samza.offset.default"));
+ assertEquals(11, generatedConfigs.size());
+ }
+
+ @Test
+ public void testSDConfigsWithoutOverrides() {
+ KafkaSystemDescriptor sd = new KafkaSystemDescriptor("kafka");
+
+ Map<String, String> generatedConfigs = sd.toConfig();
+ assertEquals("org.apache.samza.system.kafka.KafkaSystemFactory", generatedConfigs.get("systems.kafka.samza.factory"));
+ assertEquals(1, generatedConfigs.size()); // verify that there are no other configs
+ }
+}
http://git-wip-us.apache.org/repos/asf/samza/blob/2a71baf7/samza-kv/src/main/java/org/apache/samza/storage/kv/BaseLocalStoreBackedTableProvider.java
----------------------------------------------------------------------
diff --git a/samza-kv/src/main/java/org/apache/samza/storage/kv/BaseLocalStoreBackedTableProvider.java b/samza-kv/src/main/java/org/apache/samza/storage/kv/BaseLocalStoreBackedTableProvider.java
index 16da035..07f4f55 100644
--- a/samza-kv/src/main/java/org/apache/samza/storage/kv/BaseLocalStoreBackedTableProvider.java
+++ b/samza-kv/src/main/java/org/apache/samza/storage/kv/BaseLocalStoreBackedTableProvider.java
@@ -21,6 +21,7 @@ package org.apache.samza.storage.kv;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.regex.Pattern;
import org.apache.commons.lang3.StringUtils;
import org.apache.samza.SamzaException;
@@ -31,7 +32,6 @@ import org.apache.samza.config.JobConfig;
import org.apache.samza.config.MapConfig;
import org.apache.samza.config.StorageConfig;
import org.apache.samza.container.SamzaContainerContext;
-import org.apache.samza.operators.StreamGraphSpec;
import org.apache.samza.table.ReadableTable;
import org.apache.samza.table.Table;
import org.apache.samza.table.TableSpec;
@@ -50,6 +50,7 @@ import com.google.common.base.Preconditions;
* stores.
*/
abstract public class BaseLocalStoreBackedTableProvider extends BaseTableProvider {
+ public static final Pattern SYSTEM_STREAM_NAME_PATTERN = Pattern.compile("[\\d\\w-_.]+");
protected KeyValueStore kvStore;
@@ -101,8 +102,8 @@ abstract public class BaseLocalStoreBackedTableProvider extends BaseTableProvide
List<String> sideInputs = tableSpec.getSideInputs();
if (sideInputs != null && !sideInputs.isEmpty()) {
- sideInputs.forEach(si -> Preconditions.checkState(StreamGraphSpec.isValidStreamId(si), String.format(
- "Side input stream %s doesn't confirm to pattern %s", si, StreamGraphSpec.STREAM_ID_PATTERN)));
+ sideInputs.forEach(si -> Preconditions.checkState(isValidSystemStreamName(si), String.format(
+ "Side input stream %s doesn't confirm to pattern %s", si, SYSTEM_STREAM_NAME_PATTERN)));
String formattedSideInputs = String.join(",", sideInputs);
storeConfig.put(String.format(JavaStorageConfig.SIDE_INPUTS, tableSpec.getId()), formattedSideInputs);
storeConfig.put(String.format(JavaStorageConfig.SIDE_INPUTS_PROCESSOR_SERIALIZED_INSTANCE, tableSpec.getId()),
@@ -121,8 +122,8 @@ abstract public class BaseLocalStoreBackedTableProvider extends BaseTableProvide
tableSpec.getId());
}
- Preconditions.checkState(StreamGraphSpec.isValidStreamId(changelogStream), String.format(
- "Changelog stream %s doesn't confirm to pattern %s", changelogStream, StreamGraphSpec.STREAM_ID_PATTERN));
+ Preconditions.checkState(isValidSystemStreamName(changelogStream), String.format(
+ "Changelog stream %s doesn't confirm to pattern %s", changelogStream, SYSTEM_STREAM_NAME_PATTERN));
storeConfig.put(String.format(StorageConfig.CHANGELOG_STREAM(), tableSpec.getId()), changelogStream);
String changelogReplicationFactor = tableSpec.getConfig().get(
@@ -140,4 +141,8 @@ abstract public class BaseLocalStoreBackedTableProvider extends BaseTableProvide
public void close() {
logger.info("Shutting down table provider for table " + tableSpec.getId());
}
+
+ private boolean isValidSystemStreamName(String name) {
+ return StringUtils.isNotBlank(name) && SYSTEM_STREAM_NAME_PATTERN.matcher(name).matches();
+ }
}
http://git-wip-us.apache.org/repos/asf/samza/blob/2a71baf7/samza-sql/src/main/java/org/apache/samza/sql/translator/QueryTranslator.java
----------------------------------------------------------------------
diff --git a/samza-sql/src/main/java/org/apache/samza/sql/translator/QueryTranslator.java b/samza-sql/src/main/java/org/apache/samza/sql/translator/QueryTranslator.java
index 1db3000..c422130 100644
--- a/samza-sql/src/main/java/org/apache/samza/sql/translator/QueryTranslator.java
+++ b/samza-sql/src/main/java/org/apache/samza/sql/translator/QueryTranslator.java
@@ -37,8 +37,12 @@ import org.apache.samza.operators.KV;
import org.apache.samza.operators.MessageStream;
import org.apache.samza.operators.MessageStreamImpl;
import org.apache.samza.operators.StreamGraph;
+import org.apache.samza.operators.descriptors.GenericOutputDescriptor;;
+import org.apache.samza.operators.descriptors.DelegatingSystemDescriptor;
import org.apache.samza.operators.functions.MapFunction;
import org.apache.samza.operators.TableDescriptor;
+import org.apache.samza.serializers.KVSerde;
+import org.apache.samza.serializers.NoOpSerde;
import org.apache.samza.sql.data.SamzaSqlExecutionContext;
import org.apache.samza.sql.data.SamzaSqlRelMessage;
import org.apache.samza.sql.interfaces.SamzaRelConverter;
@@ -151,7 +155,11 @@ public class QueryTranslator {
Optional<TableDescriptor> tableDescriptor = sinkConfig.getTableDescriptor();
if (!tableDescriptor.isPresent()) {
- outputStream.sendTo(streamGraph.getOutputStream(sinkConfig.getStreamName()));
+ KVSerde<Object, Object> noOpKVSerde = KVSerde.of(new NoOpSerde<>(), new NoOpSerde<>());
+ String systemName = sinkConfig.getSystemName();
+ DelegatingSystemDescriptor sd = context.getSystemDescriptors().computeIfAbsent(systemName, DelegatingSystemDescriptor::new);
+ GenericOutputDescriptor<KV<Object, Object>> osd = sd.getOutputDescriptor(sinkConfig.getStreamName(), noOpKVSerde);
+ outputStream.sendTo(streamGraph.getOutputStream(osd));
} else {
Table outputTable = streamGraph.getTable(tableDescriptor.get());
if (outputTable == null) {
http://git-wip-us.apache.org/repos/asf/samza/blob/2a71baf7/samza-sql/src/main/java/org/apache/samza/sql/translator/ScanTranslator.java
----------------------------------------------------------------------
diff --git a/samza-sql/src/main/java/org/apache/samza/sql/translator/ScanTranslator.java b/samza-sql/src/main/java/org/apache/samza/sql/translator/ScanTranslator.java
index fa3d9d3..46e0840 100644
--- a/samza-sql/src/main/java/org/apache/samza/sql/translator/ScanTranslator.java
+++ b/samza-sql/src/main/java/org/apache/samza/sql/translator/ScanTranslator.java
@@ -27,7 +27,11 @@ import org.apache.samza.config.Config;
import org.apache.samza.operators.KV;
import org.apache.samza.operators.MessageStream;
import org.apache.samza.operators.StreamGraph;
+import org.apache.samza.operators.descriptors.GenericInputDescriptor;
+import org.apache.samza.operators.descriptors.DelegatingSystemDescriptor;
import org.apache.samza.operators.functions.MapFunction;
+import org.apache.samza.serializers.KVSerde;
+import org.apache.samza.serializers.NoOpSerde;
import org.apache.samza.sql.data.SamzaSqlRelMessage;
import org.apache.samza.sql.interfaces.SamzaRelConverter;
import org.apache.samza.task.TaskContext;
@@ -74,9 +78,14 @@ class ScanTranslator {
String sourceName = SqlIOConfig.getSourceFromSourceParts(tableNameParts);
Validate.isTrue(relMsgConverters.containsKey(sourceName), String.format("Unknown source %s", sourceName));
- final String streamName = systemStreamConfig.get(sourceName).getStreamName();
+ SqlIOConfig sqlIOConfig = systemStreamConfig.get(sourceName);
+ final String systemName = sqlIOConfig.getSystemName();
+ final String streamName = sqlIOConfig.getStreamName();
- MessageStream<KV<Object, Object>> inputStream = streamGraph.getInputStream(streamName);
+ KVSerde<Object, Object> noOpKVSerde = KVSerde.of(new NoOpSerde<>(), new NoOpSerde<>());
+ DelegatingSystemDescriptor sd = context.getSystemDescriptors().computeIfAbsent(systemName, DelegatingSystemDescriptor::new);
+ GenericInputDescriptor<KV<Object, Object>> isd = sd.getInputDescriptor(streamName, noOpKVSerde);
+ MessageStream<KV<Object, Object>> inputStream = streamGraph.getInputStream(isd);
MessageStream<SamzaSqlRelMessage> samzaSqlRelMessageStream = inputStream.map(new ScanMapFunction(sourceName));
context.registerMessageStream(tableScan.getId(), samzaSqlRelMessageStream);
http://git-wip-us.apache.org/repos/asf/samza/blob/2a71baf7/samza-sql/src/main/java/org/apache/samza/sql/translator/TranslatorContext.java
----------------------------------------------------------------------
diff --git a/samza-sql/src/main/java/org/apache/samza/sql/translator/TranslatorContext.java b/samza-sql/src/main/java/org/apache/samza/sql/translator/TranslatorContext.java
index fcebeef..e622d55 100644
--- a/samza-sql/src/main/java/org/apache/samza/sql/translator/TranslatorContext.java
+++ b/samza-sql/src/main/java/org/apache/samza/sql/translator/TranslatorContext.java
@@ -35,6 +35,7 @@ import org.apache.calcite.rex.RexNode;
import org.apache.calcite.schema.SchemaPlus;
import org.apache.samza.operators.MessageStream;
import org.apache.samza.operators.StreamGraph;
+import org.apache.samza.operators.descriptors.DelegatingSystemDescriptor;
import org.apache.samza.sql.data.RexToJavaCompiler;
import org.apache.samza.sql.data.SamzaSqlExecutionContext;
import org.apache.samza.sql.interfaces.SamzaRelConverter;
@@ -50,8 +51,9 @@ public class TranslatorContext implements Cloneable {
private final StreamGraph streamGraph;
private final RexToJavaCompiler compiler;
private final Map<String, SamzaRelConverter> relSamzaConverters;
- private final Map<Integer, MessageStream> messsageStreams;
+ private final Map<Integer, MessageStream> messageStreams;
private final Map<Integer, RelNode> relNodes;
+ private final Map<String, DelegatingSystemDescriptor> systemDescriptors;
/**
* The internal variables that are not shared among all cloned {@link TranslatorContext}
@@ -123,10 +125,11 @@ public class TranslatorContext implements Cloneable {
this.streamGraph = other.streamGraph;
this.compiler = other.compiler;
this.relSamzaConverters = other.relSamzaConverters;
- this.messsageStreams = other.messsageStreams;
+ this.messageStreams = other.messageStreams;
this.relNodes = other.relNodes;
this.executionContext = other.executionContext.clone();
this.dataContext = new DataContextImpl();
+ this.systemDescriptors = other.systemDescriptors;
}
/**
@@ -142,8 +145,9 @@ public class TranslatorContext implements Cloneable {
this.executionContext = executionContext;
this.dataContext = new DataContextImpl();
this.relSamzaConverters = converters;
- this.messsageStreams = new HashMap<>();
+ this.messageStreams = new HashMap<>();
this.relNodes = new HashMap<>();
+ this.systemDescriptors = new HashMap<>();
}
/**
@@ -184,7 +188,7 @@ public class TranslatorContext implements Cloneable {
* @param stream the stream
*/
void registerMessageStream(int id, MessageStream stream) {
- messsageStreams.put(id, stream);
+ messageStreams.put(id, stream);
}
/**
@@ -194,7 +198,7 @@ public class TranslatorContext implements Cloneable {
* @return the message stream
*/
MessageStream getMessageStream(int id) {
- return messsageStreams.get(id);
+ return messageStreams.get(id);
}
void registerRelNode(int id, RelNode relNode) {
@@ -209,6 +213,10 @@ public class TranslatorContext implements Cloneable {
return this.relSamzaConverters.get(source);
}
+ Map<String, DelegatingSystemDescriptor> getSystemDescriptors() {
+ return this.systemDescriptors;
+ }
+
/**
* This method helps to create a per task instance of translator context
*
http://git-wip-us.apache.org/repos/asf/samza/blob/2a71baf7/samza-sql/src/test/java/org/apache/samza/sql/translator/TestJoinTranslator.java
----------------------------------------------------------------------
diff --git a/samza-sql/src/test/java/org/apache/samza/sql/translator/TestJoinTranslator.java b/samza-sql/src/test/java/org/apache/samza/sql/translator/TestJoinTranslator.java
index 2de4856..7395a3d 100644
--- a/samza-sql/src/test/java/org/apache/samza/sql/translator/TestJoinTranslator.java
+++ b/samza-sql/src/test/java/org/apache/samza/sql/translator/TestJoinTranslator.java
@@ -147,7 +147,7 @@ public class TestJoinTranslator extends TranslatorTestBase {
when(mockOutputStream.isKeyed()).thenReturn(true);
IntermediateMessageStreamImpl
mockPartitionedStream = new IntermediateMessageStreamImpl(mockGraph, mockInputOp, mockOutputStream);
- when(mockGraph.getIntermediateStream(any(String.class), any(Serde.class))).thenReturn(mockPartitionedStream);
+ when(mockGraph.getIntermediateStream(any(String.class), any(Serde.class), eq(false))).thenReturn(mockPartitionedStream);
doAnswer(this.getRegisterMessageStreamAnswer()).when(mockContext).registerMessageStream(eq(3), any(MessageStream.class));
RexToJavaCompiler mockCompiler = mock(RexToJavaCompiler.class);
http://git-wip-us.apache.org/repos/asf/samza/blob/2a71baf7/samza-sql/src/test/java/org/apache/samza/sql/translator/TestQueryTranslator.java
----------------------------------------------------------------------
diff --git a/samza-sql/src/test/java/org/apache/samza/sql/translator/TestQueryTranslator.java b/samza-sql/src/test/java/org/apache/samza/sql/translator/TestQueryTranslator.java
index ede7995..1776067 100644
--- a/samza-sql/src/test/java/org/apache/samza/sql/translator/TestQueryTranslator.java
+++ b/samza-sql/src/test/java/org/apache/samza/sql/translator/TestQueryTranslator.java
@@ -52,7 +52,7 @@ public class TestQueryTranslator {
Assert.assertTrue(originContext.getStreamGraph() == clonedContext.getStreamGraph());
Assert.assertTrue(originContext.getExpressionCompiler() == clonedContext.getExpressionCompiler());
Assert.assertTrue(Whitebox.getInternalState(originContext, "relSamzaConverters") == Whitebox.getInternalState(clonedContext, "relSamzaConverters"));
- Assert.assertTrue(Whitebox.getInternalState(originContext, "messsageStreams") == Whitebox.getInternalState(clonedContext, "messsageStreams"));
+ Assert.assertTrue(Whitebox.getInternalState(originContext, "messageStreams") == Whitebox.getInternalState(clonedContext, "messageStreams"));
Assert.assertTrue(Whitebox.getInternalState(originContext, "relNodes") == Whitebox.getInternalState(clonedContext, "relNodes"));
Assert.assertNotEquals(originContext.getDataContext(), clonedContext.getDataContext());
validateClonedExecutionContext(originContext.getExecutionContext(), clonedContext.getExecutionContext());
@@ -137,8 +137,7 @@ public class TestQueryTranslator {
SamzaSqlApplicationConfig samzaSqlApplicationConfig = new SamzaSqlApplicationConfig(new MapConfig(config));
QueryTranslator translator = new QueryTranslator(samzaSqlApplicationConfig);
SamzaSqlQueryParser.QueryInfo queryInfo = samzaSqlApplicationConfig.getQueryInfo().get(0);
- StreamGraphSpec
- graphSpec = new StreamGraphSpec(samzaConfig);
+ StreamGraphSpec graphSpec = new StreamGraphSpec(samzaConfig);
translator.translate(queryInfo, graphSpec);
OperatorSpecGraph specGraph = graphSpec.getOperatorSpecGraph();
http://git-wip-us.apache.org/repos/asf/samza/blob/2a71baf7/samza-test/src/main/java/org/apache/samza/example/AppWithGlobalConfigExample.java
----------------------------------------------------------------------
diff --git a/samza-test/src/main/java/org/apache/samza/example/AppWithGlobalConfigExample.java b/samza-test/src/main/java/org/apache/samza/example/AppWithGlobalConfigExample.java
index c029eb4..d7f805d 100644
--- a/samza-test/src/main/java/org/apache/samza/example/AppWithGlobalConfigExample.java
+++ b/samza-test/src/main/java/org/apache/samza/example/AppWithGlobalConfigExample.java
@@ -31,6 +31,9 @@ import org.apache.samza.runtime.LocalApplicationRunner;
import org.apache.samza.serializers.JsonSerdeV2;
import org.apache.samza.serializers.KVSerde;
import org.apache.samza.serializers.StringSerde;
+import org.apache.samza.system.kafka.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.KafkaOutputDescriptor;
+import org.apache.samza.system.kafka.KafkaSystemDescriptor;
import org.apache.samza.util.CommandLine;
@@ -51,13 +54,21 @@ public class AppWithGlobalConfigExample implements StreamApplication {
@Override
public void init(StreamGraph graph, Config config) {
- graph.getInputStream("myPageViewEevent", KVSerde.of(new StringSerde("UTF-8"), new JsonSerdeV2<>(PageViewEvent.class)))
- .map(KV::getValue)
+ KafkaSystemDescriptor trackingSystem = new KafkaSystemDescriptor("tracking");
+
+ KafkaInputDescriptor<PageViewEvent> inputStreamDescriptor =
+ trackingSystem.getInputDescriptor("pageViewEvent", new JsonSerdeV2<>(PageViewEvent.class));
+
+ KafkaOutputDescriptor<KV<String, PageViewCount>> outputStreamDescriptor =
+ trackingSystem.getOutputDescriptor("pageViewEventPerMember",
+ KVSerde.of(new StringSerde(), new JsonSerdeV2<>(PageViewCount.class)));
+
+ graph.getInputStream(inputStreamDescriptor)
.window(Windows.<PageViewEvent, String, Integer>keyedTumblingWindow(m -> m.memberId, Duration.ofSeconds(10), () -> 0, (m, c) -> c + 1, null, null)
.setEarlyTrigger(Triggers.repeat(Triggers.count(5)))
.setAccumulationMode(AccumulationMode.DISCARDING), "w1")
.map(m -> KV.of(m.getKey().getKey(), new PageViewCount(m)))
- .sendTo(graph.getOutputStream("pageViewEventPerMemberStream", KVSerde.of(new StringSerde("UTF-8"), new JsonSerdeV2<>(PageViewCount.class))));
+ .sendTo(graph.getOutputStream(outputStreamDescriptor));
}
class PageViewEvent {
http://git-wip-us.apache.org/repos/asf/samza/blob/2a71baf7/samza-test/src/main/java/org/apache/samza/example/BroadcastExample.java
----------------------------------------------------------------------
diff --git a/samza-test/src/main/java/org/apache/samza/example/BroadcastExample.java b/samza-test/src/main/java/org/apache/samza/example/BroadcastExample.java
index 9ca4f35..1c1b4be 100644
--- a/samza-test/src/main/java/org/apache/samza/example/BroadcastExample.java
+++ b/samza-test/src/main/java/org/apache/samza/example/BroadcastExample.java
@@ -22,12 +22,15 @@ package org.apache.samza.example;
import org.apache.samza.application.StreamApplication;
import org.apache.samza.config.Config;
import org.apache.samza.operators.KV;
+import org.apache.samza.operators.MessageStream;
import org.apache.samza.operators.StreamGraph;
import org.apache.samza.runtime.LocalApplicationRunner;
import org.apache.samza.serializers.JsonSerdeV2;
import org.apache.samza.serializers.KVSerde;
-import org.apache.samza.operators.MessageStream;
import org.apache.samza.serializers.StringSerde;
+import org.apache.samza.system.kafka.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.KafkaOutputDescriptor;
+import org.apache.samza.system.kafka.KafkaSystemDescriptor;
import org.apache.samza.util.CommandLine;
@@ -50,12 +53,21 @@ public class BroadcastExample implements StreamApplication {
@Override
public void init(StreamGraph graph, Config config) {
- KVSerde<String, PageViewEvent> pgeMsgSerde = KVSerde.of(new StringSerde("UTF-8"), new JsonSerdeV2<>(PageViewEvent.class));
- MessageStream<KV<String, PageViewEvent>> inputStream = graph.getInputStream("pageViewEventStream", pgeMsgSerde);
+ KVSerde<String, PageViewEvent> serde = KVSerde.of(new StringSerde("UTF-8"), new JsonSerdeV2<>(PageViewEvent.class));
+ KafkaSystemDescriptor trackingSystem = new KafkaSystemDescriptor("tracking");
+ KafkaInputDescriptor<KV<String, PageViewEvent>> pageViewEvent =
+ trackingSystem.getInputDescriptor("pageViewEvent", serde);
+ KafkaOutputDescriptor<KV<String, PageViewEvent>> outStream1 =
+ trackingSystem.getOutputDescriptor("outStream1", serde);
+ KafkaOutputDescriptor<KV<String, PageViewEvent>> outStream2 =
+ trackingSystem.getOutputDescriptor("outStream2", serde);
+ KafkaOutputDescriptor<KV<String, PageViewEvent>> outStream3 =
+ trackingSystem.getOutputDescriptor("outStream3", serde);
- inputStream.filter(m -> m.key.equals("key1")).sendTo(graph.getOutputStream("outStream1", pgeMsgSerde));
- inputStream.filter(m -> m.key.equals("key2")).sendTo(graph.getOutputStream("outStream2", pgeMsgSerde));
- inputStream.filter(m -> m.key.equals("key3")).sendTo(graph.getOutputStream("outStream3", pgeMsgSerde));
+ MessageStream<KV<String, PageViewEvent>> inputStream = graph.getInputStream(pageViewEvent);
+ inputStream.filter(m -> m.key.equals("key1")).sendTo(graph.getOutputStream(outStream1));
+ inputStream.filter(m -> m.key.equals("key2")).sendTo(graph.getOutputStream(outStream2));
+ inputStream.filter(m -> m.key.equals("key3")).sendTo(graph.getOutputStream(outStream3));
}
class PageViewEvent {
http://git-wip-us.apache.org/repos/asf/samza/blob/2a71baf7/samza-test/src/main/java/org/apache/samza/example/KeyValueStoreExample.java
----------------------------------------------------------------------
diff --git a/samza-test/src/main/java/org/apache/samza/example/KeyValueStoreExample.java b/samza-test/src/main/java/org/apache/samza/example/KeyValueStoreExample.java
index 9edaabe..4d3307b 100644
--- a/samza-test/src/main/java/org/apache/samza/example/KeyValueStoreExample.java
+++ b/samza-test/src/main/java/org/apache/samza/example/KeyValueStoreExample.java
@@ -19,6 +19,10 @@
package org.apache.samza.example;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
import org.apache.samza.application.StreamApplication;
import org.apache.samza.config.Config;
import org.apache.samza.operators.KV;
@@ -31,14 +35,12 @@ import org.apache.samza.serializers.JsonSerdeV2;
import org.apache.samza.serializers.KVSerde;
import org.apache.samza.serializers.StringSerde;
import org.apache.samza.storage.kv.KeyValueStore;
+import org.apache.samza.system.kafka.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.KafkaOutputDescriptor;
+import org.apache.samza.system.kafka.KafkaSystemDescriptor;
import org.apache.samza.task.TaskContext;
import org.apache.samza.util.CommandLine;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-import java.util.concurrent.TimeUnit;
-
/**
* Example code using {@link KeyValueStore} to implement event-time window
@@ -58,12 +60,19 @@ public class KeyValueStoreExample implements StreamApplication {
@Override
public void init(StreamGraph graph, Config config) {
- MessageStream<PageViewEvent> pageViewEvents =
- graph.getInputStream("pageViewEventStream", new JsonSerdeV2<>(PageViewEvent.class));
- OutputStream<KV<String, StatsOutput>> pageViewEventPerMember =
- graph.getOutputStream("pageViewEventPerMember",
+ KafkaSystemDescriptor trackingSystem = new KafkaSystemDescriptor("tracking");
+
+ KafkaInputDescriptor<PageViewEvent> inputStreamDescriptor =
+ trackingSystem.getInputDescriptor("pageViewEvent", new JsonSerdeV2<>(PageViewEvent.class));
+
+ KafkaOutputDescriptor<KV<String, StatsOutput>> outputStreamDescriptor =
+ trackingSystem.getOutputDescriptor("pageViewEventPerMember",
KVSerde.of(new StringSerde(), new JsonSerdeV2<>(StatsOutput.class)));
+ graph.setDefaultSystem(trackingSystem);
+ MessageStream<PageViewEvent> pageViewEvents = graph.getInputStream(inputStreamDescriptor);
+ OutputStream<KV<String, StatsOutput>> pageViewEventPerMember = graph.getOutputStream(outputStreamDescriptor);
+
pageViewEvents
.partitionBy(pve -> pve.memberId, pve -> pve,
KVSerde.of(new StringSerde(), new JsonSerdeV2<>(PageViewEvent.class)), "partitionBy")
http://git-wip-us.apache.org/repos/asf/samza/blob/2a71baf7/samza-test/src/main/java/org/apache/samza/example/MergeExample.java
----------------------------------------------------------------------
diff --git a/samza-test/src/main/java/org/apache/samza/example/MergeExample.java b/samza-test/src/main/java/org/apache/samza/example/MergeExample.java
index ff983a4..33d60d6 100644
--- a/samza-test/src/main/java/org/apache/samza/example/MergeExample.java
+++ b/samza-test/src/main/java/org/apache/samza/example/MergeExample.java
@@ -20,14 +20,19 @@
package org.apache.samza.example;
import com.google.common.collect.ImmutableList;
+
import org.apache.samza.application.StreamApplication;
import org.apache.samza.config.Config;
+import org.apache.samza.operators.KV;
+import org.apache.samza.operators.MessageStream;
import org.apache.samza.operators.StreamGraph;
import org.apache.samza.runtime.LocalApplicationRunner;
import org.apache.samza.serializers.JsonSerdeV2;
import org.apache.samza.serializers.KVSerde;
-import org.apache.samza.operators.MessageStream;
import org.apache.samza.serializers.StringSerde;
+import org.apache.samza.system.kafka.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.KafkaOutputDescriptor;
+import org.apache.samza.system.kafka.KafkaSystemDescriptor;
import org.apache.samza.util.CommandLine;
public class MergeExample implements StreamApplication {
@@ -45,14 +50,22 @@ public class MergeExample implements StreamApplication {
@Override
public void init(StreamGraph graph, Config config) {
+ KVSerde<String, PageViewEvent> serde = KVSerde.of(new StringSerde("UTF-8"), new JsonSerdeV2<>(PageViewEvent.class));
+ KafkaSystemDescriptor trackingSystem = new KafkaSystemDescriptor("tracking");
- KVSerde<String, PageViewEvent>
- pgeMsgSerde = KVSerde.of(new StringSerde("UTF-8"), new JsonSerdeV2<>(PageViewEvent.class));
+ KafkaInputDescriptor<KV<String, PageViewEvent>> isd1 =
+ trackingSystem.getInputDescriptor("pageViewStream1", serde);
+ KafkaInputDescriptor<KV<String, PageViewEvent>> isd2 =
+ trackingSystem.getInputDescriptor("pageViewStream2", serde);
+ KafkaInputDescriptor<KV<String, PageViewEvent>> isd3 =
+ trackingSystem.getInputDescriptor("pageViewStream3", serde);
- MessageStream.mergeAll(ImmutableList.of(graph.getInputStream("viewStream1", pgeMsgSerde),
- graph.getInputStream("viewStream2", pgeMsgSerde), graph.getInputStream("viewStream3", pgeMsgSerde)))
- .sendTo(graph.getOutputStream("mergedStream", pgeMsgSerde));
+ KafkaOutputDescriptor<KV<String, PageViewEvent>> osd =
+ trackingSystem.getOutputDescriptor("mergedStream", serde);
+ MessageStream
+ .mergeAll(ImmutableList.of(graph.getInputStream(isd1), graph.getInputStream(isd2), graph.getInputStream(isd3)))
+ .sendTo(graph.getOutputStream(osd));
}
class PageViewEvent {
http://git-wip-us.apache.org/repos/asf/samza/blob/2a71baf7/samza-test/src/main/java/org/apache/samza/example/OrderShipmentJoinExample.java
----------------------------------------------------------------------
diff --git a/samza-test/src/main/java/org/apache/samza/example/OrderShipmentJoinExample.java b/samza-test/src/main/java/org/apache/samza/example/OrderShipmentJoinExample.java
index 1c0bc25..34b5fc6 100644
--- a/samza-test/src/main/java/org/apache/samza/example/OrderShipmentJoinExample.java
+++ b/samza-test/src/main/java/org/apache/samza/example/OrderShipmentJoinExample.java
@@ -18,21 +18,21 @@
*/
package org.apache.samza.example;
+import java.time.Duration;
import org.apache.samza.application.StreamApplication;
import org.apache.samza.config.Config;
import org.apache.samza.operators.KV;
-import org.apache.samza.operators.MessageStream;
-import org.apache.samza.operators.OutputStream;
import org.apache.samza.operators.StreamGraph;
import org.apache.samza.operators.functions.JoinFunction;
import org.apache.samza.runtime.LocalApplicationRunner;
import org.apache.samza.serializers.JsonSerdeV2;
import org.apache.samza.serializers.KVSerde;
import org.apache.samza.serializers.StringSerde;
+import org.apache.samza.system.kafka.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.KafkaOutputDescriptor;
+import org.apache.samza.system.kafka.KafkaSystemDescriptor;
import org.apache.samza.util.CommandLine;
-import java.time.Duration;
-
/**
* Simple 2-way stream-to-stream join example
@@ -52,21 +52,22 @@ public class OrderShipmentJoinExample implements StreamApplication {
@Override
public void init(StreamGraph graph, Config config) {
-
- MessageStream<OrderRecord> orders =
- graph.getInputStream("orders", new JsonSerdeV2<>(OrderRecord.class));
- MessageStream<ShipmentRecord> shipments =
- graph.getInputStream("shipments", new JsonSerdeV2<>(ShipmentRecord.class));
- OutputStream<KV<String, FulfilledOrderRecord>> fulfilledOrders =
- graph.getOutputStream("fulfilledOrders",
+ KafkaSystemDescriptor trackingSystem = new KafkaSystemDescriptor("tracking");
+
+ KafkaInputDescriptor<OrderRecord> orderStreamDescriptor =
+ trackingSystem.getInputDescriptor("orders", new JsonSerdeV2<>(OrderRecord.class));
+ KafkaInputDescriptor<ShipmentRecord> shipmentStreamDescriptor =
+ trackingSystem.getInputDescriptor("shipments", new JsonSerdeV2<>(ShipmentRecord.class));
+ KafkaOutputDescriptor<KV<String, FulfilledOrderRecord>> fulfilledOrdersStreamDescriptor =
+ trackingSystem.getOutputDescriptor("fulfilledOrders",
KVSerde.of(new StringSerde(), new JsonSerdeV2<>(FulfilledOrderRecord.class)));
- orders
- .join(shipments, new MyJoinFunction(),
+ graph.getInputStream(orderStreamDescriptor)
+ .join(graph.getInputStream(shipmentStreamDescriptor), new MyJoinFunction(),
new StringSerde(), new JsonSerdeV2<>(OrderRecord.class), new JsonSerdeV2<>(ShipmentRecord.class),
Duration.ofMinutes(1), "join")
.map(fulFilledOrder -> KV.of(fulFilledOrder.orderId, fulFilledOrder))
- .sendTo(fulfilledOrders);
+ .sendTo(graph.getOutputStream(fulfilledOrdersStreamDescriptor));
}
http://git-wip-us.apache.org/repos/asf/samza/blob/2a71baf7/samza-test/src/main/java/org/apache/samza/example/PageViewCounterExample.java
----------------------------------------------------------------------
diff --git a/samza-test/src/main/java/org/apache/samza/example/PageViewCounterExample.java b/samza-test/src/main/java/org/apache/samza/example/PageViewCounterExample.java
index 2581506..dc5eb74 100644
--- a/samza-test/src/main/java/org/apache/samza/example/PageViewCounterExample.java
+++ b/samza-test/src/main/java/org/apache/samza/example/PageViewCounterExample.java
@@ -35,6 +35,9 @@ import org.apache.samza.runtime.LocalApplicationRunner;
import org.apache.samza.serializers.JsonSerdeV2;
import org.apache.samza.serializers.KVSerde;
import org.apache.samza.serializers.StringSerde;
+import org.apache.samza.system.kafka.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.KafkaOutputDescriptor;
+import org.apache.samza.system.kafka.KafkaSystemDescriptor;
import org.apache.samza.util.CommandLine;
@@ -56,13 +59,18 @@ public class PageViewCounterExample implements StreamApplication {
@Override
public void init(StreamGraph graph, Config config) {
+ KafkaSystemDescriptor trackingSystem = new KafkaSystemDescriptor("tracking");
- MessageStream<PageViewEvent> pageViewEvents = null;
- pageViewEvents = graph.getInputStream("pageViewEventStream", new JsonSerdeV2<>(PageViewEvent.class));
- OutputStream<KV<String, PageViewCount>> pageViewEventPerMemberStream =
- graph.getOutputStream("pageViewEventPerMemberStream",
+ KafkaInputDescriptor<PageViewEvent> inputStreamDescriptor =
+ trackingSystem.getInputDescriptor("pageViewEvent", new JsonSerdeV2<>(PageViewEvent.class));
+
+ KafkaOutputDescriptor<KV<String, PageViewCount>> outputStreamDescriptor =
+ trackingSystem.getOutputDescriptor("pageViewEventPerMember",
KVSerde.of(new StringSerde(), new JsonSerdeV2<>(PageViewCount.class)));
+ MessageStream<PageViewEvent> pageViewEvents = graph.getInputStream(inputStreamDescriptor);
+ OutputStream<KV<String, PageViewCount>> pageViewEventPerMemberStream = graph.getOutputStream(outputStreamDescriptor);
+
SupplierFunction<Integer> initialValue = () -> 0;
FoldLeftFunction<PageViewEvent, Integer> foldLeftFn = (m, c) -> c + 1;
pageViewEvents
@@ -71,7 +79,6 @@ public class PageViewCounterExample implements StreamApplication {
.setAccumulationMode(AccumulationMode.DISCARDING), "tumblingWindow")
.map(windowPane -> KV.of(windowPane.getKey().getKey(), new PageViewCount(windowPane)))
.sendTo(pageViewEventPerMemberStream);
-
}
class PageViewEvent {
http://git-wip-us.apache.org/repos/asf/samza/blob/2a71baf7/samza-test/src/main/java/org/apache/samza/example/RepartitionExample.java
----------------------------------------------------------------------
diff --git a/samza-test/src/main/java/org/apache/samza/example/RepartitionExample.java b/samza-test/src/main/java/org/apache/samza/example/RepartitionExample.java
index 7f28346..b776c7d 100644
--- a/samza-test/src/main/java/org/apache/samza/example/RepartitionExample.java
+++ b/samza-test/src/main/java/org/apache/samza/example/RepartitionExample.java
@@ -18,6 +18,7 @@
*/
package org.apache.samza.example;
+import java.time.Duration;
import org.apache.samza.application.StreamApplication;
import org.apache.samza.config.Config;
import org.apache.samza.operators.KV;
@@ -30,10 +31,11 @@ import org.apache.samza.runtime.LocalApplicationRunner;
import org.apache.samza.serializers.JsonSerdeV2;
import org.apache.samza.serializers.KVSerde;
import org.apache.samza.serializers.StringSerde;
+import org.apache.samza.system.kafka.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.KafkaOutputDescriptor;
+import org.apache.samza.system.kafka.KafkaSystemDescriptor;
import org.apache.samza.util.CommandLine;
-import java.time.Duration;
-
/**
* Example {@link StreamApplication} code to test the API methods with re-partition operator
@@ -53,18 +55,24 @@ public class RepartitionExample implements StreamApplication {
@Override
public void init(StreamGraph graph, Config config) {
+ KafkaSystemDescriptor trackingSystem = new KafkaSystemDescriptor("tracking");
- MessageStream<PageViewEvent> pageViewEvents =
- graph.getInputStream("pageViewEvent", new JsonSerdeV2<>(PageViewEvent.class));
- OutputStream<KV<String, MyStreamOutput>> pageViewEventPerMember =
- graph.getOutputStream("pageViewEventPerMember",
+ KafkaInputDescriptor<PageViewEvent> inputStreamDescriptor =
+ trackingSystem.getInputDescriptor("pageViewEvent", new JsonSerdeV2<>(PageViewEvent.class));
+
+ KafkaOutputDescriptor<KV<String, MyStreamOutput>> outputStreamDescriptor =
+ trackingSystem.getOutputDescriptor("pageViewEventPerMember",
KVSerde.of(new StringSerde(), new JsonSerdeV2<>(MyStreamOutput.class)));
+ graph.setDefaultSystem(trackingSystem);
+ MessageStream<PageViewEvent> pageViewEvents = graph.getInputStream(inputStreamDescriptor);
+ OutputStream<KV<String, MyStreamOutput>> pageViewEventPerMember = graph.getOutputStream(outputStreamDescriptor);
+
pageViewEvents
.partitionBy(pve -> pve.memberId, pve -> pve,
KVSerde.of(new StringSerde(), new JsonSerdeV2<>(PageViewEvent.class)), "partitionBy")
- .window(Windows.keyedTumblingWindow(KV::getKey, Duration.ofMinutes(5), () -> 0, (m, c) -> c + 1, null, null),
- "window")
+ .window(Windows.keyedTumblingWindow(
+ KV::getKey, Duration.ofMinutes(5), () -> 0, (m, c) -> c + 1, null, null), "window")
.map(windowPane -> KV.of(windowPane.getKey().getKey(), new MyStreamOutput(windowPane)))
.sendTo(pageViewEventPerMember);
http://git-wip-us.apache.org/repos/asf/samza/blob/2a71baf7/samza-test/src/main/java/org/apache/samza/example/WindowExample.java
----------------------------------------------------------------------
diff --git a/samza-test/src/main/java/org/apache/samza/example/WindowExample.java b/samza-test/src/main/java/org/apache/samza/example/WindowExample.java
index 4950695..cbc1e8e 100644
--- a/samza-test/src/main/java/org/apache/samza/example/WindowExample.java
+++ b/samza-test/src/main/java/org/apache/samza/example/WindowExample.java
@@ -19,6 +19,7 @@
package org.apache.samza.example;
+import java.time.Duration;
import org.apache.samza.application.StreamApplication;
import org.apache.samza.config.Config;
import org.apache.samza.operators.MessageStream;
@@ -26,16 +27,17 @@ import org.apache.samza.operators.OutputStream;
import org.apache.samza.operators.StreamGraph;
import org.apache.samza.operators.functions.FoldLeftFunction;
import org.apache.samza.operators.functions.SupplierFunction;
-import org.apache.samza.operators.windows.WindowPane;
import org.apache.samza.operators.triggers.Triggers;
+import org.apache.samza.operators.windows.WindowPane;
import org.apache.samza.operators.windows.Windows;
import org.apache.samza.runtime.LocalApplicationRunner;
import org.apache.samza.serializers.IntegerSerde;
import org.apache.samza.serializers.JsonSerdeV2;
+import org.apache.samza.system.kafka.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.KafkaOutputDescriptor;
+import org.apache.samza.system.kafka.KafkaSystemDescriptor;
import org.apache.samza.util.CommandLine;
-import java.time.Duration;
-
/**
* Example implementation of a simple user-defined task w/ a window operator.
@@ -56,11 +58,18 @@ public class WindowExample implements StreamApplication {
@Override
public void init(StreamGraph graph, Config config) {
+ KafkaSystemDescriptor trackingSystem = new KafkaSystemDescriptor("tracking");
+
+ KafkaInputDescriptor<PageViewEvent> inputStreamDescriptor =
+ trackingSystem.getInputDescriptor("pageViewEvent", new JsonSerdeV2<>(PageViewEvent.class));
+ KafkaOutputDescriptor<Integer> outputStreamDescriptor =
+ trackingSystem.getOutputDescriptor("pageViewEventPerMember", new IntegerSerde());
+
+ MessageStream<PageViewEvent> inputStream = graph.getInputStream(inputStreamDescriptor);
+ OutputStream<Integer> outputStream = graph.getOutputStream(outputStreamDescriptor);
SupplierFunction<Integer> initialValue = () -> 0;
FoldLeftFunction<PageViewEvent, Integer> counter = (m, c) -> c == null ? 1 : c + 1;
- MessageStream<PageViewEvent> inputStream = graph.getInputStream("inputStream", new JsonSerdeV2<PageViewEvent>());
- OutputStream<Integer> outputStream = graph.getOutputStream("outputStream", new IntegerSerde());
// create a tumbling window that outputs the number of message collected every 10 minutes.
// also emit early results if either the number of messages collected reaches 30000, or if no new messages arrive
http://git-wip-us.apache.org/repos/asf/samza/blob/2a71baf7/samza-test/src/main/java/org/apache/samza/test/framework/TestRunner.java
----------------------------------------------------------------------
diff --git a/samza-test/src/main/java/org/apache/samza/test/framework/TestRunner.java b/samza-test/src/main/java/org/apache/samza/test/framework/TestRunner.java
index 3c45967..5ca497a 100644
--- a/samza-test/src/main/java/org/apache/samza/test/framework/TestRunner.java
+++ b/samza-test/src/main/java/org/apache/samza/test/framework/TestRunner.java
@@ -138,7 +138,7 @@ public class TestRunner {
*/
private void registerSystem(String systemName) {
if (!systems.containsKey(systemName)) {
- systems.put(systemName, CollectionStreamSystemSpec.create(systemName));
+ systems.put(systemName, CollectionStreamSystemSpec.create(systemName, JOB_NAME));
configs.putAll(systems.get(systemName).getSystemConfigs());
}
}
http://git-wip-us.apache.org/repos/asf/samza/blob/2a71baf7/samza-test/src/main/java/org/apache/samza/test/framework/system/CollectionStreamSystemSpec.java
----------------------------------------------------------------------
diff --git a/samza-test/src/main/java/org/apache/samza/test/framework/system/CollectionStreamSystemSpec.java b/samza-test/src/main/java/org/apache/samza/test/framework/system/CollectionStreamSystemSpec.java
index c005c41..5658f61 100644
--- a/samza-test/src/main/java/org/apache/samza/test/framework/system/CollectionStreamSystemSpec.java
+++ b/samza-test/src/main/java/org/apache/samza/test/framework/system/CollectionStreamSystemSpec.java
@@ -22,6 +22,7 @@ package org.apache.samza.test.framework.system;
import com.google.common.base.Preconditions;
import java.util.HashMap;
import java.util.Map;
+import org.apache.commons.lang3.StringUtils;
import org.apache.samza.system.inmemory.InMemorySystemFactory;
@@ -32,11 +33,17 @@ import org.apache.samza.system.inmemory.InMemorySystemFactory;
* Following system level configs are set by default
* <ol>
* <li>"systems.%s.default.stream.samza.offset.default" = "oldest"</li>
+ * <li>"jobs.job-name.systems.%s.default.stream.samza.offset.default" = "oldest"</li>
* <li>"systems.%s.samza.factory" = {@link InMemorySystemFactory}</li>
+ * <li>"jobs.job-name.systems.%s.samza.factory" = {@link InMemorySystemFactory}</li>
* </ol>
- *
+ * The "systems.*" configs are required since the planner uses the system to get metadata about streams during
+ * planning. The "jobs.job-name.systems.*" configs are required since configs generated from user provided
+ * system/stream descriptors override configs originally supplied to the planner. Configs in the "jobs.job-name.*"
+ * scope have the highest precedence.
*/
public class CollectionStreamSystemSpec {
+ private static final String CONFIG_OVERRIDE_PREFIX = "jobs.%s."; // prefix to override configs generated by the planner
private static final String SYSTEM_FACTORY = "systems.%s.samza.factory";
private static final String SYSTEM_OFFSET = "systems.%s.default.stream.samza.offset.default";
@@ -51,11 +58,13 @@ public class CollectionStreamSystemSpec {
* <p>
* @param systemName represents unique name of the system
*/
- private CollectionStreamSystemSpec(String systemName) {
+ private CollectionStreamSystemSpec(String systemName, String jobName) {
this.systemName = systemName;
systemConfigs = new HashMap<String, String>();
systemConfigs.put(String.format(SYSTEM_FACTORY, systemName), InMemorySystemFactory.class.getName());
+ systemConfigs.put(String.format(CONFIG_OVERRIDE_PREFIX + SYSTEM_FACTORY, jobName, systemName), InMemorySystemFactory.class.getName());
systemConfigs.put(String.format(SYSTEM_OFFSET, systemName), "oldest");
+ systemConfigs.put(String.format(CONFIG_OVERRIDE_PREFIX + SYSTEM_OFFSET, jobName, systemName), "oldest");
}
public String getSystemName() {
@@ -67,13 +76,15 @@ public class CollectionStreamSystemSpec {
}
/**
- * Creates a {@link CollectionStreamSystemSpec} with name {@code name}
- * @param name represents name of the {@link CollectionStreamSystemSpec}
+ * Creates a {@link CollectionStreamSystemSpec} with name {@code systemName}
+ * @param systemName represents name of the {@link CollectionStreamSystemSpec}
+ * @param jobName name of the job
* @return an instance of {@link CollectionStreamSystemSpec}
*/
- public static CollectionStreamSystemSpec create(String name) {
- Preconditions.checkState(name != null);
- return new CollectionStreamSystemSpec(name);
+ public static CollectionStreamSystemSpec create(String systemName, String jobName) {
+ Preconditions.checkState(StringUtils.isNotBlank(systemName));
+ Preconditions.checkState(StringUtils.isNotBlank(jobName));
+ return new CollectionStreamSystemSpec(systemName, jobName);
}
}
http://git-wip-us.apache.org/repos/asf/samza/blob/2a71baf7/samza-test/src/main/java/org/apache/samza/test/integration/TestStandaloneIntegrationApplication.java
----------------------------------------------------------------------
diff --git a/samza-test/src/main/java/org/apache/samza/test/integration/TestStandaloneIntegrationApplication.java b/samza-test/src/main/java/org/apache/samza/test/integration/TestStandaloneIntegrationApplication.java
index f6e3d5f..af20fd7 100644
--- a/samza-test/src/main/java/org/apache/samza/test/integration/TestStandaloneIntegrationApplication.java
+++ b/samza-test/src/main/java/org/apache/samza/test/integration/TestStandaloneIntegrationApplication.java
@@ -20,7 +20,13 @@ package org.apache.samza.test.integration;
import org.apache.samza.application.StreamApplication;
import org.apache.samza.config.Config;
+import org.apache.samza.operators.KV;
import org.apache.samza.operators.StreamGraph;
+import org.apache.samza.serializers.KVSerde;
+import org.apache.samza.serializers.NoOpSerde;
+import org.apache.samza.system.kafka.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.KafkaOutputDescriptor;
+import org.apache.samza.system.kafka.KafkaSystemDescriptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -29,14 +35,21 @@ import org.slf4j.LoggerFactory;
* Acts as a pass through filter for all the events from a input stream.
*/
public class TestStandaloneIntegrationApplication implements StreamApplication {
-
private static final Logger LOGGER = LoggerFactory.getLogger(TestStandaloneIntegrationApplication.class);
@Override
public void init(StreamGraph graph, Config config) {
- String inputStream = config.get("input.stream.name");
+ String systemName = "testSystemName";
+ String inputStreamName = config.get("input.stream.name");
String outputStreamName = "standaloneIntegrationTestKafkaOutputTopic";
- LOGGER.info("Publishing message to: {}.", outputStreamName);
- graph.getInputStream(inputStream).sendTo(graph.getOutputStream(outputStreamName));
+ LOGGER.info("Publishing message from: {} to: {}.", inputStreamName, outputStreamName);
+ KafkaSystemDescriptor kafkaSystemDescriptor = new KafkaSystemDescriptor(systemName);
+
+ KVSerde<Object, Object> noOpSerde = KVSerde.of(new NoOpSerde<>(), new NoOpSerde<>());
+ KafkaInputDescriptor<KV<Object, Object>> isd =
+ kafkaSystemDescriptor.getInputDescriptor(inputStreamName, noOpSerde);
+ KafkaOutputDescriptor<KV<Object, Object>> osd =
+ kafkaSystemDescriptor.getOutputDescriptor(inputStreamName, noOpSerde);
+ graph.getInputStream(isd).sendTo(graph.getOutputStream(osd));
}
}
http://git-wip-us.apache.org/repos/asf/samza/blob/2a71baf7/samza-test/src/test/java/org/apache/samza/test/controlmessages/EndOfStreamIntegrationTest.java
----------------------------------------------------------------------
diff --git a/samza-test/src/test/java/org/apache/samza/test/controlmessages/EndOfStreamIntegrationTest.java b/samza-test/src/test/java/org/apache/samza/test/controlmessages/EndOfStreamIntegrationTest.java
index 3301af8..b86c6af 100644
--- a/samza-test/src/test/java/org/apache/samza/test/controlmessages/EndOfStreamIntegrationTest.java
+++ b/samza-test/src/test/java/org/apache/samza/test/controlmessages/EndOfStreamIntegrationTest.java
@@ -31,8 +31,12 @@ import org.apache.samza.config.MapConfig;
import org.apache.samza.config.TaskConfig;
import org.apache.samza.container.grouper.task.SingleContainerGrouperFactory;
import org.apache.samza.operators.KV;
+import org.apache.samza.operators.descriptors.GenericInputDescriptor;
+import org.apache.samza.operators.descriptors.DelegatingSystemDescriptor;
import org.apache.samza.operators.functions.MapFunction;
import org.apache.samza.runtime.LocalApplicationRunner;
+import org.apache.samza.serializers.KVSerde;
+import org.apache.samza.serializers.NoOpSerde;
import org.apache.samza.standalone.PassthroughCoordinationUtilsFactory;
import org.apache.samza.standalone.PassthroughJobCoordinatorFactory;
import org.apache.samza.test.controlmessages.TestData.PageView;
@@ -40,8 +44,8 @@ import org.apache.samza.test.controlmessages.TestData.PageViewJsonSerdeFactory;
import org.apache.samza.test.harness.AbstractIntegrationTestHarness;
import org.apache.samza.test.util.ArraySystemFactory;
import org.apache.samza.test.util.Base64Serializer;
-
import org.junit.Test;
+
import static org.junit.Assert.assertEquals;
/**
@@ -92,7 +96,10 @@ public class EndOfStreamIntegrationTest extends AbstractIntegrationTestHarness {
final LocalApplicationRunner runner = new LocalApplicationRunner(new MapConfig(configs));
final StreamApplication app = (streamGraph, cfg) -> {
- streamGraph.<KV<String, PageView>>getInputStream("PageView")
+ DelegatingSystemDescriptor sd = new DelegatingSystemDescriptor("test");
+ GenericInputDescriptor<KV<String, PageView>> isd =
+ sd.getInputDescriptor("PageView", KVSerde.of(new NoOpSerde<>(), new NoOpSerde<>()));
+ streamGraph.getInputStream(isd)
.map(Values.create())
.partitionBy(pv -> pv.getMemberId(), pv -> pv, "p1")
.sink((m, collector, coordinator) -> {
http://git-wip-us.apache.org/repos/asf/samza/blob/2a71baf7/samza-test/src/test/java/org/apache/samza/test/controlmessages/WatermarkIntegrationTest.java
----------------------------------------------------------------------
diff --git a/samza-test/src/test/java/org/apache/samza/test/controlmessages/WatermarkIntegrationTest.java b/samza-test/src/test/java/org/apache/samza/test/controlmessages/WatermarkIntegrationTest.java
index d4dc4ed..5336595 100644
--- a/samza-test/src/test/java/org/apache/samza/test/controlmessages/WatermarkIntegrationTest.java
+++ b/samza-test/src/test/java/org/apache/samza/test/controlmessages/WatermarkIntegrationTest.java
@@ -19,6 +19,8 @@
package org.apache.samza.test.controlmessages;
+import scala.collection.JavaConverters;
+
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.HashMap;
@@ -38,6 +40,8 @@ import org.apache.samza.container.TaskName;
import org.apache.samza.container.grouper.task.SingleContainerGrouperFactory;
import org.apache.samza.metrics.MetricsRegistry;
import org.apache.samza.operators.KV;
+import org.apache.samza.operators.descriptors.GenericInputDescriptor;
+import org.apache.samza.operators.descriptors.DelegatingSystemDescriptor;
import org.apache.samza.operators.impl.InputOperatorImpl;
import org.apache.samza.operators.impl.OperatorImpl;
import org.apache.samza.operators.impl.OperatorImplGraph;
@@ -48,6 +52,8 @@ import org.apache.samza.processor.TestStreamProcessorUtil;
import org.apache.samza.runtime.LocalApplicationRunner;
import org.apache.samza.runtime.TestLocalApplicationRunner;
import org.apache.samza.serializers.IntegerSerdeFactory;
+import org.apache.samza.serializers.KVSerde;
+import org.apache.samza.serializers.NoOpSerde;
import org.apache.samza.serializers.StringSerdeFactory;
import org.apache.samza.standalone.PassthroughCoordinationUtilsFactory;
import org.apache.samza.standalone.PassthroughJobCoordinatorFactory;
@@ -66,7 +72,6 @@ import org.apache.samza.test.harness.AbstractIntegrationTestHarness;
import org.apache.samza.test.util.SimpleSystemAdmin;
import org.apache.samza.test.util.TestStreamConsumer;
import org.junit.Test;
-import scala.collection.JavaConverters;
import static org.junit.Assert.assertEquals;
@@ -143,7 +148,10 @@ public class WatermarkIntegrationTest extends AbstractIntegrationTestHarness {
List<PageView> received = new ArrayList<>();
final StreamApplication app = (streamGraph, cfg) -> {
- streamGraph.<KV<String, PageView>>getInputStream("PageView")
+ DelegatingSystemDescriptor sd = new DelegatingSystemDescriptor("test");
+ GenericInputDescriptor<KV<String, PageView>> isd =
+ sd.getInputDescriptor("PageView", KVSerde.of(new NoOpSerde<>(), new NoOpSerde<>()));
+ streamGraph.getInputStream(isd)
.map(EndOfStreamIntegrationTest.Values.create())
.partitionBy(pv -> pv.getMemberId(), pv -> pv, "p1")
.sink((m, collector, coordinator) -> {
http://git-wip-us.apache.org/repos/asf/samza/blob/2a71baf7/samza-test/src/test/java/org/apache/samza/test/framework/BroadcastAssertApp.java
----------------------------------------------------------------------
diff --git a/samza-test/src/test/java/org/apache/samza/test/framework/BroadcastAssertApp.java b/samza-test/src/test/java/org/apache/samza/test/framework/BroadcastAssertApp.java
index ec52aa4..aca6c40 100644
--- a/samza-test/src/test/java/org/apache/samza/test/framework/BroadcastAssertApp.java
+++ b/samza-test/src/test/java/org/apache/samza/test/framework/BroadcastAssertApp.java
@@ -19,17 +19,18 @@
package org.apache.samza.test.framework;
+import java.util.Arrays;
import org.apache.samza.application.StreamApplication;
import org.apache.samza.config.Config;
import org.apache.samza.operators.MessageStream;
import org.apache.samza.operators.StreamGraph;
import org.apache.samza.serializers.JsonSerdeV2;
+import org.apache.samza.system.kafka.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.KafkaSystemDescriptor;
import org.apache.samza.test.operator.data.PageView;
-import java.util.Arrays;
-
public class BroadcastAssertApp implements StreamApplication {
-
+ public static final String SYSTEM = "kafka";
public static final String INPUT_TOPIC_NAME_PROP = "inputTopicName";
@@ -38,8 +39,10 @@ public class BroadcastAssertApp implements StreamApplication {
String inputTopic = config.get(INPUT_TOPIC_NAME_PROP);
final JsonSerdeV2<PageView> serde = new JsonSerdeV2<>(PageView.class);
+ KafkaSystemDescriptor ksd = new KafkaSystemDescriptor(SYSTEM);
+ KafkaInputDescriptor<PageView> isd = ksd.getInputDescriptor(inputTopic, serde);
final MessageStream<PageView> broadcastPageViews = graph
- .getInputStream(inputTopic, serde)
+ .getInputStream(isd)
.broadcast(serde, "pv");
/**
http://git-wip-us.apache.org/repos/asf/samza/blob/2a71baf7/samza-test/src/test/java/org/apache/samza/test/framework/StreamApplicationIntegrationTest.java
----------------------------------------------------------------------
diff --git a/samza-test/src/test/java/org/apache/samza/test/framework/StreamApplicationIntegrationTest.java b/samza-test/src/test/java/org/apache/samza/test/framework/StreamApplicationIntegrationTest.java
index ba4c985..6fdf887 100644
--- a/samza-test/src/test/java/org/apache/samza/test/framework/StreamApplicationIntegrationTest.java
+++ b/samza-test/src/test/java/org/apache/samza/test/framework/StreamApplicationIntegrationTest.java
@@ -27,27 +27,40 @@ import java.util.Random;
import org.apache.samza.SamzaException;
import org.apache.samza.application.StreamApplication;
import org.apache.samza.operators.KV;
+import org.apache.samza.operators.MessageStream;
import org.apache.samza.operators.functions.MapFunction;
+import org.apache.samza.serializers.KVSerde;
+import org.apache.samza.serializers.NoOpSerde;
import org.apache.samza.system.OutgoingMessageEnvelope;
import org.apache.samza.system.SystemStream;
+import org.apache.samza.system.kafka.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.KafkaSystemDescriptor;
import org.apache.samza.test.controlmessages.TestData;
import org.apache.samza.test.framework.stream.CollectionStream;
-import static org.apache.samza.test.controlmessages.TestData.PageView;
import org.junit.Assert;
import org.junit.Test;
+import static org.apache.samza.test.controlmessages.TestData.PageView;
+
public class StreamApplicationIntegrationTest {
final StreamApplication pageViewFilter = (streamGraph, cfg) -> {
- streamGraph.<KV<String, TestData.PageView>>getInputStream("PageView").map(
- StreamApplicationIntegrationTest.Values.create()).filter(pv -> pv.getPageKey().equals("inbox"));
+ KafkaSystemDescriptor ksd = new KafkaSystemDescriptor("test");
+ KafkaInputDescriptor<KV<String, PageView>> isd =
+ ksd.getInputDescriptor("PageView", KVSerde.of(new NoOpSerde<>(), new NoOpSerde<>()));
+ MessageStream<KV<String, TestData.PageView>> inputStream = streamGraph.getInputStream(isd);
+ inputStream.map(StreamApplicationIntegrationTest.Values.create()).filter(pv -> pv.getPageKey().equals("inbox"));
};
- final StreamApplication pageViewParition = (streamGraph, cfg) -> {
- streamGraph.<KV<String, PageView>>getInputStream("PageView")
+ final StreamApplication pageViewRepartition = (streamGraph, cfg) -> {
+ KafkaSystemDescriptor ksd = new KafkaSystemDescriptor("test");
+ KafkaInputDescriptor<KV<String, PageView>> isd =
+ ksd.getInputDescriptor("PageView", KVSerde.of(new NoOpSerde<>(), new NoOpSerde<>()));
+ MessageStream<KV<String, TestData.PageView>> inputStream = streamGraph.getInputStream(isd);
+ inputStream
.map(Values.create())
- .partitionBy(pv -> pv.getMemberId(), pv -> pv, "p1")
+ .partitionBy(PageView::getMemberId, pv -> pv, "p1")
.sink((m, collector, coordinator) -> {
collector.send(new OutgoingMessageEnvelope(new SystemStream("test", "Output"),
m.getKey(), m.getKey(),
@@ -74,7 +87,7 @@ public class StreamApplicationIntegrationTest {
CollectionStream output = CollectionStream.empty("test", "Output", 10);
TestRunner
- .of(pageViewParition)
+ .of(pageViewRepartition)
.addInputStream(input)
.addOutputStream(output)
.addOverrideConfig("job.default.system", "test")
@@ -99,7 +112,7 @@ public class StreamApplicationIntegrationTest {
CollectionStream output = CollectionStream.empty("test", "Output", 10);
TestRunner
- .of(pageViewParition)
+ .of(pageViewRepartition)
.addInputStream(input)
.addOutputStream(output)
.run(Duration.ofMillis(1000));