Consolidating package names for System, Stream, Application and Table descriptors.
authorPrateek Maheshwari <pmaheshwari@apache.org>
Sat, 13 Oct 2018 01:34:34 +0000 (18:34 -0700)
committerPrateek Maheshwari <pmaheshwari@apache.org>
Sat, 13 Oct 2018 01:34:34 +0000 (18:34 -0700)
Everything in this PR is either:
1. A package name change and a corresponding file move.
2. Javadoc changes to use FQN in link tags to fix checkstyle complaints about unused imports, and corresponding fixes to make them fit within line width. No change in contents.
3. In a couple of places, changing method visibility to public with VisibleForTesting annotations for accessing them from Tests (RemoteReadWriteTable.java, RemoteReadableTable.java)

Author: Prateek Maheshwari <pmaheshwari@apache.org>

Reviewers: Bharath Kumarasubramanian <bkumaras@linkedin.com>, Jagadish Venkatraman <vjagadish1989@gmail.com>, Yi Pan <nickpan47@gmail.com>

Closes #720 from prateekm/descriptor-package-cleanup

182 files changed:
docs/startup/quick-start/versioned/index.md
samza-api/src/main/java/org/apache/samza/application/SamzaApplication.java
samza-api/src/main/java/org/apache/samza/application/StreamApplication.java
samza-api/src/main/java/org/apache/samza/application/TaskApplication.java
samza-api/src/main/java/org/apache/samza/application/descriptors/ApplicationDescriptor.java [moved from samza-api/src/main/java/org/apache/samza/application/ApplicationDescriptor.java with 91% similarity]
samza-api/src/main/java/org/apache/samza/application/descriptors/StreamApplicationDescriptor.java [moved from samza-api/src/main/java/org/apache/samza/application/StreamApplicationDescriptor.java with 92% similarity]
samza-api/src/main/java/org/apache/samza/application/descriptors/TaskApplicationDescriptor.java [moved from samza-api/src/main/java/org/apache/samza/application/TaskApplicationDescriptor.java with 84% similarity]
samza-api/src/main/java/org/apache/samza/context/ApplicationContainerContextFactory.java
samza-api/src/main/java/org/apache/samza/context/ApplicationTaskContextFactory.java
samza-api/src/main/java/org/apache/samza/operators/MessageStream.java
samza-api/src/main/java/org/apache/samza/operators/functions/ClosableFunction.java
samza-api/src/main/java/org/apache/samza/operators/functions/InitableFunction.java
samza-api/src/main/java/org/apache/samza/system/descriptors/ExpandingInputDescriptorProvider.java [moved from samza-api/src/main/java/org/apache/samza/operators/descriptors/base/system/ExpandingInputDescriptorProvider.java with 93% similarity]
samza-api/src/main/java/org/apache/samza/system/descriptors/GenericInputDescriptor.java [moved from samza-api/src/main/java/org/apache/samza/operators/descriptors/GenericInputDescriptor.java with 88% similarity]
samza-api/src/main/java/org/apache/samza/system/descriptors/GenericOutputDescriptor.java [moved from samza-api/src/main/java/org/apache/samza/operators/descriptors/GenericOutputDescriptor.java with 88% similarity]
samza-api/src/main/java/org/apache/samza/system/descriptors/GenericSystemDescriptor.java [moved from samza-api/src/main/java/org/apache/samza/operators/descriptors/GenericSystemDescriptor.java with 89% similarity]
samza-api/src/main/java/org/apache/samza/system/descriptors/InputDescriptor.java [moved from samza-api/src/main/java/org/apache/samza/operators/descriptors/base/stream/InputDescriptor.java with 97% similarity]
samza-api/src/main/java/org/apache/samza/system/descriptors/InputTransformer.java [moved from samza-api/src/main/java/org/apache/samza/operators/functions/InputTransformer.java with 91% similarity]
samza-api/src/main/java/org/apache/samza/system/descriptors/OutputDescriptor.java [moved from samza-api/src/main/java/org/apache/samza/operators/descriptors/base/stream/OutputDescriptor.java with 92% similarity]
samza-api/src/main/java/org/apache/samza/system/descriptors/OutputDescriptorProvider.java [moved from samza-api/src/main/java/org/apache/samza/operators/descriptors/base/system/OutputDescriptorProvider.java with 94% similarity]
samza-api/src/main/java/org/apache/samza/system/descriptors/SimpleInputDescriptorProvider.java [moved from samza-api/src/main/java/org/apache/samza/operators/descriptors/base/system/SimpleInputDescriptorProvider.java with 92% similarity]
samza-api/src/main/java/org/apache/samza/system/descriptors/StreamDescriptor.java [moved from samza-api/src/main/java/org/apache/samza/operators/descriptors/base/stream/StreamDescriptor.java with 97% similarity]
samza-api/src/main/java/org/apache/samza/system/descriptors/StreamExpander.java [moved from samza-api/src/main/java/org/apache/samza/operators/functions/StreamExpander.java with 93% similarity]
samza-api/src/main/java/org/apache/samza/system/descriptors/SystemDescriptor.java [moved from samza-api/src/main/java/org/apache/samza/operators/descriptors/base/system/SystemDescriptor.java with 96% similarity]
samza-api/src/main/java/org/apache/samza/system/descriptors/TransformingInputDescriptorProvider.java [moved from samza-api/src/main/java/org/apache/samza/operators/descriptors/base/system/TransformingInputDescriptorProvider.java with 93% similarity]
samza-api/src/main/java/org/apache/samza/table/TableDescriptorsProvider.java
samza-api/src/main/java/org/apache/samza/table/TableSpec.java
samza-api/src/main/java/org/apache/samza/table/descriptors/TableDescriptor.java [moved from samza-api/src/main/java/org/apache/samza/operators/TableDescriptor.java with 98% similarity]
samza-api/src/main/java/org/apache/samza/table/descriptors/TableProvider.java [moved from samza-api/src/main/java/org/apache/samza/table/TableProvider.java with 96% similarity]
samza-api/src/main/java/org/apache/samza/table/descriptors/TableProviderFactory.java [moved from samza-api/src/main/java/org/apache/samza/table/TableProviderFactory.java with 93% similarity]
samza-api/src/test/java/org/apache/samza/system/descriptors/TestExpandingInputDescriptor.java [moved from samza-api/src/test/java/org/apache/samza/operators/descriptors/TestExpandingInputDescriptor.java with 87% similarity]
samza-api/src/test/java/org/apache/samza/system/descriptors/TestGenericInputDescriptor.java [moved from samza-api/src/test/java/org/apache/samza/operators/descriptors/TestGenericInputDescriptor.java with 99% similarity]
samza-api/src/test/java/org/apache/samza/system/descriptors/TestGenericSystemDescriptor.java [moved from samza-api/src/test/java/org/apache/samza/operators/descriptors/TestGenericSystemDescriptor.java with 98% similarity]
samza-api/src/test/java/org/apache/samza/system/descriptors/TestSimpleInputDescriptor.java [moved from samza-api/src/test/java/org/apache/samza/operators/descriptors/TestSimpleInputDescriptor.java with 88% similarity]
samza-api/src/test/java/org/apache/samza/system/descriptors/TestTransformingInputDescriptor.java [moved from samza-api/src/test/java/org/apache/samza/operators/descriptors/TestTransformingInputDescriptor.java with 87% similarity]
samza-api/src/test/java/org/apache/samza/system/descriptors/examples/expanding/ExampleExpandingInputDescriptor.java [moved from samza-api/src/test/java/org/apache/samza/operators/descriptors/expanding/ExampleExpandingInputDescriptor.java with 81% similarity]
samza-api/src/test/java/org/apache/samza/system/descriptors/examples/expanding/ExampleExpandingOutputDescriptor.java [moved from samza-api/src/test/java/org/apache/samza/operators/descriptors/expanding/ExampleExpandingOutputDescriptor.java with 85% similarity]
samza-api/src/test/java/org/apache/samza/system/descriptors/examples/expanding/ExampleExpandingSystemDescriptor.java [moved from samza-api/src/test/java/org/apache/samza/operators/descriptors/expanding/ExampleExpandingSystemDescriptor.java with 84% similarity]
samza-api/src/test/java/org/apache/samza/system/descriptors/examples/serde/ExampleSimpleInputDescriptor.java [moved from samza-api/src/test/java/org/apache/samza/operators/descriptors/serde/ExampleSimpleInputDescriptor.java with 82% similarity]
samza-api/src/test/java/org/apache/samza/system/descriptors/examples/serde/ExampleSimpleOutputDescriptor.java [moved from samza-api/src/test/java/org/apache/samza/operators/descriptors/serde/ExampleSimpleOutputDescriptor.java with 85% similarity]
samza-api/src/test/java/org/apache/samza/system/descriptors/examples/serde/ExampleSimpleSystemDescriptor.java [moved from samza-api/src/test/java/org/apache/samza/operators/descriptors/serde/ExampleSimpleSystemDescriptor.java with 84% similarity]
samza-api/src/test/java/org/apache/samza/system/descriptors/examples/transforming/ExampleTransformingInputDescriptor.java [moved from samza-api/src/test/java/org/apache/samza/operators/descriptors/transforming/ExampleTransformingInputDescriptor.java with 81% similarity]
samza-api/src/test/java/org/apache/samza/system/descriptors/examples/transforming/ExampleTransformingOutputDescriptor.java [moved from samza-api/src/test/java/org/apache/samza/operators/descriptors/transforming/ExampleTransformingOutputDescriptor.java with 84% similarity]
samza-api/src/test/java/org/apache/samza/system/descriptors/examples/transforming/ExampleTransformingSystemDescriptor.java [moved from samza-api/src/test/java/org/apache/samza/operators/descriptors/transforming/ExampleTransformingSystemDescriptor.java with 84% similarity]
samza-azure/src/main/java/org/apache/samza/system/eventhub/descriptors/EventHubsInputDescriptor.java [moved from samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubsInputDescriptor.java with 95% similarity]
samza-azure/src/main/java/org/apache/samza/system/eventhub/descriptors/EventHubsOutputDescriptor.java [moved from samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubsOutputDescriptor.java with 94% similarity]
samza-azure/src/main/java/org/apache/samza/system/eventhub/descriptors/EventHubsSystemDescriptor.java [moved from samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubsSystemDescriptor.java with 97% similarity]
samza-azure/src/test/java/org/apache/samza/system/eventhub/descriptors/TestEventHubsInputDescriptor.java [moved from samza-azure/src/test/java/org/apache/samza/system/eventhub/TestEventHubsInputDescriptor.java with 97% similarity]
samza-azure/src/test/java/org/apache/samza/system/eventhub/descriptors/TestEventHubsOutputDescriptor.java [moved from samza-azure/src/test/java/org/apache/samza/system/eventhub/TestEventHubsOutputDescriptor.java with 97% similarity]
samza-azure/src/test/java/org/apache/samza/system/eventhub/descriptors/TestEventHubsSystemDescriptor.java [moved from samza-azure/src/test/java/org/apache/samza/system/eventhub/TestEventHubsSystemDescriptor.java with 98% similarity]
samza-core/src/main/java/org/apache/samza/application/LegacyTaskApplication.java
samza-core/src/main/java/org/apache/samza/application/descriptors/ApplicationDescriptorImpl.java [moved from samza-core/src/main/java/org/apache/samza/application/ApplicationDescriptorImpl.java with 96% similarity]
samza-core/src/main/java/org/apache/samza/application/descriptors/ApplicationDescriptorUtil.java [moved from samza-core/src/main/java/org/apache/samza/application/ApplicationDescriptorUtil.java with 90% similarity]
samza-core/src/main/java/org/apache/samza/application/descriptors/StreamApplicationDescriptorImpl.java [moved from samza-core/src/main/java/org/apache/samza/application/StreamApplicationDescriptorImpl.java with 96% similarity]
samza-core/src/main/java/org/apache/samza/application/descriptors/TaskApplicationDescriptorImpl.java [moved from samza-core/src/main/java/org/apache/samza/application/TaskApplicationDescriptorImpl.java with 93% similarity]
samza-core/src/main/java/org/apache/samza/config/JavaTableConfig.java
samza-core/src/main/java/org/apache/samza/execution/ExecutionPlanner.java
samza-core/src/main/java/org/apache/samza/execution/JobGraph.java
samza-core/src/main/java/org/apache/samza/execution/JobGraphJsonGenerator.java
samza-core/src/main/java/org/apache/samza/execution/JobNode.java
samza-core/src/main/java/org/apache/samza/execution/JobPlanner.java
samza-core/src/main/java/org/apache/samza/execution/LocalJobPlanner.java
samza-core/src/main/java/org/apache/samza/execution/RemoteJobPlanner.java
samza-core/src/main/java/org/apache/samza/operators/MessageStreamImpl.java
samza-core/src/main/java/org/apache/samza/operators/OperatorSpecGraph.java
samza-core/src/main/java/org/apache/samza/operators/impl/InputOperatorImpl.java
samza-core/src/main/java/org/apache/samza/operators/spec/InputOperatorSpec.java
samza-core/src/main/java/org/apache/samza/operators/spec/OperatorSpec.java
samza-core/src/main/java/org/apache/samza/operators/spec/OperatorSpecs.java
samza-core/src/main/java/org/apache/samza/operators/stream/IntermediateMessageStreamImpl.java
samza-core/src/main/java/org/apache/samza/runtime/LocalApplicationRunner.java
samza-core/src/main/java/org/apache/samza/runtime/LocalContainerRunner.java
samza-core/src/main/java/org/apache/samza/runtime/RemoteApplicationRunner.java
samza-core/src/main/java/org/apache/samza/system/descriptors/DelegatingSystemDescriptor.java [moved from samza-core/src/main/java/org/apache/samza/operators/descriptors/DelegatingSystemDescriptor.java with 88% similarity]
samza-core/src/main/java/org/apache/samza/table/TableConfigGenerator.java
samza-core/src/main/java/org/apache/samza/table/TableManager.java
samza-core/src/main/java/org/apache/samza/table/caching/descriptors/CachingTableDescriptor.java [moved from samza-core/src/main/java/org/apache/samza/table/caching/CachingTableDescriptor.java with 90% similarity]
samza-core/src/main/java/org/apache/samza/table/caching/descriptors/CachingTableProvider.java [moved from samza-core/src/main/java/org/apache/samza/table/caching/CachingTableProvider.java with 95% similarity]
samza-core/src/main/java/org/apache/samza/table/caching/descriptors/CachingTableProviderFactory.java [moved from samza-core/src/main/java/org/apache/samza/table/caching/CachingTableProviderFactory.java with 80% similarity]
samza-core/src/main/java/org/apache/samza/table/caching/guava/descriptors/GuavaCacheTableDescriptor.java [moved from samza-core/src/main/java/org/apache/samza/table/caching/guava/GuavaCacheTableDescriptor.java with 91% similarity]
samza-core/src/main/java/org/apache/samza/table/caching/guava/descriptors/GuavaCacheTableProvider.java [moved from samza-core/src/main/java/org/apache/samza/table/caching/guava/GuavaCacheTableProvider.java with 90% similarity]
samza-core/src/main/java/org/apache/samza/table/caching/guava/descriptors/GuavaCacheTableProviderFactory.java [moved from samza-core/src/main/java/org/apache/samza/table/caching/guava/GuavaCacheTableProviderFactory.java with 79% similarity]
samza-core/src/main/java/org/apache/samza/table/descriptors/BaseHybridTableDescriptor.java [moved from samza-core/src/main/java/org/apache/samza/table/hybrid/BaseHybridTableDescriptor.java with 91% similarity]
samza-core/src/main/java/org/apache/samza/table/descriptors/BaseTableDescriptor.java [moved from samza-core/src/main/java/org/apache/samza/operators/BaseTableDescriptor.java with 98% similarity]
samza-core/src/main/java/org/apache/samza/table/remote/RemoteReadWriteTable.java
samza-core/src/main/java/org/apache/samza/table/remote/RemoteReadableTable.java
samza-core/src/main/java/org/apache/samza/table/remote/descriptors/RemoteTableDescriptor.java [moved from samza-core/src/main/java/org/apache/samza/table/remote/RemoteTableDescriptor.java with 97% similarity]
samza-core/src/main/java/org/apache/samza/table/remote/descriptors/RemoteTableProvider.java [moved from samza-core/src/main/java/org/apache/samza/table/remote/RemoteTableProvider.java with 92% similarity]
samza-core/src/main/java/org/apache/samza/table/remote/descriptors/RemoteTableProviderFactory.java [moved from samza-core/src/main/java/org/apache/samza/table/remote/RemoteTableProviderFactory.java with 87% similarity]
samza-core/src/main/java/org/apache/samza/table/retry/RetriableReadFunction.java
samza-core/src/main/java/org/apache/samza/table/retry/RetriableWriteFunction.java
samza-core/src/main/java/org/apache/samza/table/utils/descriptors/BaseTableProvider.java [moved from samza-core/src/main/java/org/apache/samza/table/utils/BaseTableProvider.java with 95% similarity]
samza-core/src/main/java/org/apache/samza/task/StreamOperatorTask.java
samza-core/src/main/java/org/apache/samza/task/TaskFactoryUtil.java
samza-core/src/main/scala/org/apache/samza/job/local/ThreadJobFactory.scala
samza-core/src/test/java/org/apache/samza/application/MockStreamApplication.java
samza-core/src/test/java/org/apache/samza/application/TestApplicationUtil.java
samza-core/src/test/java/org/apache/samza/application/descriptors/TestStreamApplicationDescriptorImpl.java [moved from samza-core/src/test/java/org/apache/samza/application/TestStreamApplicationDescriptorImpl.java with 97% similarity]
samza-core/src/test/java/org/apache/samza/application/descriptors/TestTaskApplicationDescriptorImpl.java [moved from samza-core/src/test/java/org/apache/samza/application/TestTaskApplicationDescriptorImpl.java with 93% similarity]
samza-core/src/test/java/org/apache/samza/execution/ExecutionPlannerTestBase.java
samza-core/src/test/java/org/apache/samza/execution/TestExecutionPlanner.java
samza-core/src/test/java/org/apache/samza/execution/TestJobGraph.java
samza-core/src/test/java/org/apache/samza/execution/TestJobGraphJsonGenerator.java
samza-core/src/test/java/org/apache/samza/execution/TestJobNodeConfigurationGenerator.java
samza-core/src/test/java/org/apache/samza/execution/TestLocalJobPlanner.java
samza-core/src/test/java/org/apache/samza/execution/TestRemoteJobPlanner.java
samza-core/src/test/java/org/apache/samza/operators/TestJoinOperator.java
samza-core/src/test/java/org/apache/samza/operators/TestMessageStreamImpl.java
samza-core/src/test/java/org/apache/samza/operators/TestOperatorSpecGraph.java
samza-core/src/test/java/org/apache/samza/operators/impl/TestOperatorImplGraph.java
samza-core/src/test/java/org/apache/samza/operators/impl/TestWindowOperator.java
samza-core/src/test/java/org/apache/samza/operators/spec/TestPartitionByOperatorSpec.java
samza-core/src/test/java/org/apache/samza/runtime/TestLocalApplicationRunner.java
samza-core/src/test/java/org/apache/samza/table/TestTableManager.java
samza-core/src/test/java/org/apache/samza/table/caching/TestCachingTable.java
samza-core/src/test/java/org/apache/samza/table/remote/descriptors/TestRemoteTableDescriptor.java [moved from samza-core/src/test/java/org/apache/samza/table/remote/TestRemoteTableDescriptor.java with 89% similarity]
samza-core/src/test/java/org/apache/samza/task/TestTaskFactoryUtil.java
samza-kafka/src/main/java/org/apache/samza/system/kafka/descriptors/KafkaInputDescriptor.java [moved from samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaInputDescriptor.java with 95% similarity]
samza-kafka/src/main/java/org/apache/samza/system/kafka/descriptors/KafkaOutputDescriptor.java [moved from samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaOutputDescriptor.java with 88% similarity]
samza-kafka/src/main/java/org/apache/samza/system/kafka/descriptors/KafkaSystemDescriptor.java [moved from samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemDescriptor.java with 97% similarity]
samza-kafka/src/test/java/org/apache/samza/system/kafka/TestKafkaSystemAdminWithMock.java
samza-kafka/src/test/java/org/apache/samza/system/kafka/descriptors/TestKafkaInputDescriptor.java [moved from samza-kafka/src/test/java/org/apache/samza/system/kafka/TestKafkaInputDescriptor.java with 91% similarity]
samza-kafka/src/test/java/org/apache/samza/system/kafka/descriptors/TestKafkaSystemDescriptor.java [moved from samza-kafka/src/test/java/org/apache/samza/system/kafka/TestKafkaSystemDescriptor.java with 96% similarity]
samza-kv-inmemory/src/main/java/org/apache/samza/storage/kv/inmemory/descriptors/InMemoryTableDescriptor.java [moved from samza-kv-inmemory/src/main/java/org/apache/samza/storage/kv/inmemory/InMemoryTableDescriptor.java with 94% similarity]
samza-kv-inmemory/src/main/java/org/apache/samza/storage/kv/inmemory/descriptors/InMemoryTableProvider.java [moved from samza-kv-inmemory/src/main/java/org/apache/samza/storage/kv/inmemory/InMemoryTableProvider.java with 91% similarity]
samza-kv-inmemory/src/main/java/org/apache/samza/storage/kv/inmemory/descriptors/InMemoryTableProviderFactory.java [moved from samza-kv-inmemory/src/main/java/org/apache/samza/storage/kv/inmemory/InMemoryTableProviderFactory.java with 86% similarity]
samza-kv-inmemory/src/test/java/org/apache/samza/storage/kv/inmemory/descriptors/TestInMemoryTableDescriptor.java [moved from samza-kv-inmemory/src/test/java/org/apache/samza/storage/kv/inmemory/TestInMemoryTableDescriptor.java with 96% similarity]
samza-kv-inmemory/src/test/java/org/apache/samza/storage/kv/inmemory/descriptors/TestInMemoryTableProvider.java [moved from samza-kv-inmemory/src/test/java/org/apache/samza/storage/kv/inmemory/TestInMemoryTableProvider.java with 93% similarity]
samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/descriptors/RocksDbTableDescriptor.java [moved from samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/RocksDbTableDescriptor.java with 99% similarity]
samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/descriptors/RocksDbTableProvider.java [moved from samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/RocksDbTableProvider.java with 95% similarity]
samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/descriptors/RocksDbTableProviderFactory.java [moved from samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/RocksDbTableProviderFactory.java with 86% similarity]
samza-kv-rocksdb/src/test/java/org/apache/samza/storage/kv/descriptors/TestRocksDbTableDescriptor.java [moved from samza-kv-rocksdb/src/test/java/org/apache/samza/storage/kv/TestRocksDbTableDescriptor.java with 98% similarity]
samza-kv-rocksdb/src/test/java/org/apache/samza/storage/kv/descriptors/TestRocksDbTableProvider.java [moved from samza-kv-rocksdb/src/test/java/org/apache/samza/storage/kv/TestRocksDbTableProvider.java with 93% similarity]
samza-kv/src/main/java/org/apache/samza/storage/kv/descriptors/BaseLocalStoreBackedTableDescriptor.java [moved from samza-kv/src/main/java/org/apache/samza/storage/kv/BaseLocalStoreBackedTableDescriptor.java with 98% similarity]
samza-kv/src/main/java/org/apache/samza/storage/kv/descriptors/BaseLocalStoreBackedTableProvider.java [moved from samza-kv/src/main/java/org/apache/samza/storage/kv/BaseLocalStoreBackedTableProvider.java with 96% similarity]
samza-kv/src/test/java/org/apache/samza/storage/kv/descriptors/TestBaseLocalStoreBackedTableProvider.java [moved from samza-kv/src/test/java/org/apache/samza/storage/kv/TestBaseLocalStoreBackedTableProvider.java with 97% similarity]
samza-sql/src/main/java/org/apache/samza/sql/impl/ConfigBasedIOResolverFactory.java
samza-sql/src/main/java/org/apache/samza/sql/interfaces/SqlIOConfig.java
samza-sql/src/main/java/org/apache/samza/sql/runner/SamzaSqlApplication.java
samza-sql/src/main/java/org/apache/samza/sql/translator/ModifyTranslator.java
samza-sql/src/main/java/org/apache/samza/sql/translator/QueryTranslator.java
samza-sql/src/main/java/org/apache/samza/sql/translator/ScanTranslator.java
samza-sql/src/main/java/org/apache/samza/sql/translator/TranslatorContext.java
samza-sql/src/test/java/org/apache/samza/sql/testutil/TestIOResolverFactory.java
samza-sql/src/test/java/org/apache/samza/sql/translator/TestFilterTranslator.java
samza-sql/src/test/java/org/apache/samza/sql/translator/TestJoinTranslator.java
samza-sql/src/test/java/org/apache/samza/sql/translator/TestProjectTranslator.java
samza-sql/src/test/java/org/apache/samza/sql/translator/TestQueryTranslator.java
samza-sql/src/test/java/org/apache/samza/sql/translator/TranslatorTestBase.java
samza-test/src/main/java/org/apache/samza/example/AppWithGlobalConfigExample.java
samza-test/src/main/java/org/apache/samza/example/BroadcastExample.java
samza-test/src/main/java/org/apache/samza/example/KeyValueStoreExample.java
samza-test/src/main/java/org/apache/samza/example/MergeExample.java
samza-test/src/main/java/org/apache/samza/example/OrderShipmentJoinExample.java
samza-test/src/main/java/org/apache/samza/example/PageViewCounterExample.java
samza-test/src/main/java/org/apache/samza/example/RepartitionExample.java
samza-test/src/main/java/org/apache/samza/example/TaskApplicationExample.java
samza-test/src/main/java/org/apache/samza/example/WindowExample.java
samza-test/src/main/java/org/apache/samza/test/framework/StreamAssert.java
samza-test/src/main/java/org/apache/samza/test/framework/TestRunner.java
samza-test/src/main/java/org/apache/samza/test/framework/system/descriptors/InMemoryInputDescriptor.java [moved from samza-test/src/main/java/org/apache/samza/test/framework/system/InMemoryInputDescriptor.java with 93% similarity]
samza-test/src/main/java/org/apache/samza/test/framework/system/descriptors/InMemoryOutputDescriptor.java [moved from samza-test/src/main/java/org/apache/samza/test/framework/system/InMemoryOutputDescriptor.java with 89% similarity]
samza-test/src/main/java/org/apache/samza/test/framework/system/descriptors/InMemorySystemDescriptor.java [moved from samza-test/src/main/java/org/apache/samza/test/framework/system/InMemorySystemDescriptor.java with 93% similarity]
samza-test/src/main/java/org/apache/samza/test/integration/TestStandaloneIntegrationApplication.java
samza-test/src/test/java/org/apache/samza/test/controlmessages/EndOfStreamIntegrationTest.java
samza-test/src/test/java/org/apache/samza/test/controlmessages/WatermarkIntegrationTest.java
samza-test/src/test/java/org/apache/samza/test/framework/AsyncStreamTaskIntegrationTest.java
samza-test/src/test/java/org/apache/samza/test/framework/BroadcastAssertApp.java
samza-test/src/test/java/org/apache/samza/test/framework/FaultInjectionTest.java
samza-test/src/test/java/org/apache/samza/test/framework/StreamApplicationIntegrationTest.java
samza-test/src/test/java/org/apache/samza/test/framework/StreamTaskIntegrationTest.java
samza-test/src/test/java/org/apache/samza/test/framework/TestSchedulingApp.java
samza-test/src/test/java/org/apache/samza/test/operator/RepartitionJoinWindowApp.java
samza-test/src/test/java/org/apache/samza/test/operator/RepartitionWindowApp.java
samza-test/src/test/java/org/apache/samza/test/operator/SessionWindowApp.java
samza-test/src/test/java/org/apache/samza/test/operator/TumblingWindowApp.java
samza-test/src/test/java/org/apache/samza/test/processor/TestStreamApplication.java
samza-test/src/test/java/org/apache/samza/test/table/TestLocalTable.java
samza-test/src/test/java/org/apache/samza/test/table/TestLocalTableWithSideInputs.java
samza-test/src/test/java/org/apache/samza/test/table/TestRemoteTable.java
samza-test/src/test/java/org/apache/samza/test/table/TestTableDescriptorsProvider.java
samza-tools/src/main/java/org/apache/samza/tools/benchmark/SystemConsumerWithSamzaBench.java

index 44b8376..a046ee7 100644 (file)
@@ -54,7 +54,7 @@ Now let’s write some code! The first step is to create your own Samza applicat
 package samzaapp;
 
 import org.apache.samza.application.StreamApplication;
-import org.apache.samza.application.StreamApplicationDescriptor;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 
 public class WordCount implements StreamApplication {
  @Override
index 7606be8..5423e2e 100644 (file)
@@ -19,6 +19,7 @@
 package org.apache.samza.application;
 
 import org.apache.samza.annotation.InterfaceStability;
+import org.apache.samza.application.descriptors.ApplicationDescriptor;
 
 
 /**
index a83cb37..fe77045 100644 (file)
@@ -19,6 +19,7 @@
 package org.apache.samza.application;
 
 import org.apache.samza.annotation.InterfaceStability;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 
 /**
  * Describes and initializes the transforms for processing message streams and generating results in high-level API. 
index 424634d..d84aa12 100644 (file)
@@ -19,6 +19,7 @@
 package org.apache.samza.application;
 
 import org.apache.samza.annotation.InterfaceStability;
+import org.apache.samza.application.descriptors.TaskApplicationDescriptor;
 
 
 /**
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.application;
+package org.apache.samza.application.descriptors;
 
 import java.util.Map;
 import org.apache.samza.annotation.InterfaceStability;
@@ -28,10 +28,12 @@ import org.apache.samza.runtime.ProcessorLifecycleListenerFactory;
 
 
 /**
- * The interface class to describe the configuration, input and output streams, and processing logic in a {@link SamzaApplication}.
+ * The interface class to describe the configuration, input and output streams, and processing logic in a
+ * {@link org.apache.samza.application.SamzaApplication}.
  * <p>
- * Sub-classes {@link StreamApplicationDescriptor} and {@link TaskApplicationDescriptor} are specific interfaces for applications
- * written in high-level {@link StreamApplication} and low-level {@link TaskApplication} APIs, respectively.
+ * Sub-classes {@link StreamApplicationDescriptor} and {@link TaskApplicationDescriptor} are specific interfaces for
+ * applications written in high-level {@link org.apache.samza.application.StreamApplication} and low-level
+ * {@link org.apache.samza.application.TaskApplication} APIs, respectively.
  *
  * @param <S> sub-class of user application descriptor.
  */
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.application;
+package org.apache.samza.application.descriptors;
 
 import org.apache.samza.annotation.InterfaceStability;
+import org.apache.samza.system.descriptors.InputDescriptor;
+import org.apache.samza.system.descriptors.OutputDescriptor;
 import org.apache.samza.operators.KV;
 import org.apache.samza.operators.MessageStream;
 import org.apache.samza.operators.OutputStream;
-import org.apache.samza.operators.TableDescriptor;
-import org.apache.samza.operators.descriptors.base.stream.InputDescriptor;
-import org.apache.samza.operators.descriptors.base.stream.OutputDescriptor;
-import org.apache.samza.operators.descriptors.base.system.SystemDescriptor;
+import org.apache.samza.table.descriptors.TableDescriptor;
+import org.apache.samza.system.descriptors.SystemDescriptor;
 import org.apache.samza.table.Table;
 
 
 /**
- * The interface class to describe a {@link SamzaApplication} in high-level API in Samza.
+ * The interface class to describe a {@link org.apache.samza.application.SamzaApplication} in high-level API in Samza.
  */
 @InterfaceStability.Evolving
 public interface StreamApplicationDescriptor extends ApplicationDescriptor<StreamApplicationDescriptor> {
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.application;
+package org.apache.samza.application.descriptors;
 
 import org.apache.samza.annotation.InterfaceStability;
-import org.apache.samza.operators.TableDescriptor;
-import org.apache.samza.operators.descriptors.base.stream.InputDescriptor;
-import org.apache.samza.operators.descriptors.base.stream.OutputDescriptor;
+import org.apache.samza.system.descriptors.InputDescriptor;
+import org.apache.samza.system.descriptors.OutputDescriptor;
+import org.apache.samza.table.descriptors.TableDescriptor;
 import org.apache.samza.task.TaskFactory;
 
 
 /**
- *  The interface to describe a {@link SamzaApplication} that uses low-level API task for processing.
+ *  The interface to describe a {@link org.apache.samza.application.SamzaApplication} that uses low-level API task
+ *  for processing.
  */
 @InterfaceStability.Evolving
 public interface TaskApplicationDescriptor extends ApplicationDescriptor<TaskApplicationDescriptor> {
index fbc2eef..074b0b4 100644 (file)
@@ -29,7 +29,8 @@ import java.io.Serializable;
  * stage. At that stage, the framework-provided job-level and container-level contexts are available for creating the
  * {@link ApplicationContainerContext}.
  * <p>
- * This is {@link Serializable} because it is specified in {@link org.apache.samza.application.ApplicationDescriptor}.
+ * This is {@link Serializable} because it is specified in the
+ * {@link org.apache.samza.application.descriptors.ApplicationDescriptor}.
  * @param <T> concrete type of {@link ApplicationContainerContext} returned by this factory
  */
 public interface ApplicationContainerContextFactory<T extends ApplicationContainerContext> extends Serializable {
index af9ad68..619bbc7 100644 (file)
@@ -29,7 +29,8 @@ import java.io.Serializable;
  * task. At that stage, the framework-provided job-level, container-level, and task-level contexts are available for
  * creating the {@link ApplicationTaskContext}. Also, the application-defined container-level context is available.
  * <p>
- * This is {@link Serializable} because it is specified in {@link org.apache.samza.application.ApplicationDescriptor}.
+ * This is {@link Serializable} because it is specified in the
+ * {@link org.apache.samza.application.descriptors.ApplicationDescriptor}.
  * @param <T> concrete type of {@link ApplicationTaskContext} returned by this factory
  */
 public interface ApplicationTaskContextFactory<T extends ApplicationTaskContext> extends Serializable {
index 97ac65d..f951a84 100644 (file)
@@ -40,7 +40,7 @@ import org.apache.samza.table.Table;
  * A stream of messages that can be transformed into another {@link MessageStream}.
  * <p>
  * A {@link MessageStream} corresponding to an input stream can be obtained using
- * {@link org.apache.samza.application.StreamApplicationDescriptor#getInputStream}.
+ * {@link org.apache.samza.application.descriptors.StreamApplicationDescriptor#getInputStream}.
  *
  * @param <M> the type of messages in this stream
  */
@@ -214,7 +214,8 @@ public interface MessageStream<M> {
 
   /**
    * Re-partitions this {@link MessageStream} using keys from the {@code keyExtractor} by creating a new
-   * intermediate stream on the default system provided via {@link org.apache.samza.application.StreamApplicationDescriptor#withDefaultSystem}.
+   * intermediate stream on the default system provided via
+   * {@link org.apache.samza.application.descriptors.StreamApplicationDescriptor#withDefaultSystem}.
    * This intermediate stream is both an output and input to the job.
    * <p>
    * Uses the provided {@link KVSerde} for serialization of keys and values.
index 12823cc..d3c1a23 100644 (file)
@@ -24,12 +24,12 @@ import org.apache.samza.annotation.InterfaceStability;
 
 /**
  * A function that can be closed after its execution.
- *
- * <p> Implement {@link #close()} to free resources used during the execution of the function, clean up state etc.
- *
- * <p> Order of finalization: {@link ClosableFunction}s are closed in the reverse topological order of operators in the
- * {@link org.apache.samza.application.StreamApplicationDescriptor}. For any two operators A and B in the graph, if operator B
- * consumes results from operator A, then operator B is guaranteed to be closed before operator A.
+ * <p>
+ * Implement {@link #close()} to free resources used during the execution of the function, clean up state etc.
+ * <p>
+ * Order of finalization: {@link ClosableFunction}s are closed in the reverse topological order of operators in the
+ * {@link org.apache.samza.application.descriptors.StreamApplicationDescriptor}. For any two operators A and B in the
+ * graph, if operator B consumes results from operator A, then operator B is guaranteed to be closed before operator A.
  *
  */
 @InterfaceStability.Evolving
index 7f950de..c808876 100644 (file)
@@ -24,10 +24,11 @@ import org.apache.samza.context.Context;
 
 /**
  * A function that can be initialized before execution.
- *
- * <p> Order of initialization: {@link InitableFunction}s are invoked in the topological order of operators in the
- * {@link org.apache.samza.application.StreamApplicationDescriptor}. For any two operators A and B in the graph, if operator B
- * consumes results from operator A, then operator A is guaranteed to be initialized before operator B.
+ * <p>
+ * Order of initialization: {@link InitableFunction}s are invoked in the topological order of operators in the
+ * {@link org.apache.samza.application.descriptors.StreamApplicationDescriptor}. For any two operators A and B in the
+ * graph, if operator B consumes results from operator A, then operator A is guaranteed to be initialized before
+ * operator B.
  *
  */
 @InterfaceStability.Evolving
@@ -16,9 +16,8 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.descriptors.base.system;
+package org.apache.samza.system.descriptors;
 
-import org.apache.samza.operators.descriptors.base.stream.InputDescriptor;
 import org.apache.samza.serializers.Serde;
 
 /**
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.descriptors;
+package org.apache.samza.system.descriptors;
 
-import org.apache.samza.operators.descriptors.base.stream.InputDescriptor;
-import org.apache.samza.operators.descriptors.base.system.SystemDescriptor;
 import org.apache.samza.serializers.Serde;
 
 /**
  * A descriptor for a generic input stream.
  * <p>
- * An instance of this descriptor may be obtained from an appropriately configured {@link GenericSystemDescriptor}.
+ * An instance of this descriptor may be obtained from an appropriately configured
+ * {@link GenericSystemDescriptor}.
  * <p>
  * If the system being used provides its own system and stream descriptor implementations, they should be used instead.
  * Otherwise, this {@link GenericInputDescriptor} may be used to provide Samza-specific properties of the input stream.
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.descriptors;
+package org.apache.samza.system.descriptors;
 
-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 generic output stream.
  * <p>
- * An instance of this descriptor may be obtained from an appropriately configured {@link GenericSystemDescriptor}.
+ * An instance of this descriptor may be obtained from an appropriately configured
+ * {@link GenericSystemDescriptor}.
  * <p>
  * If the system being used provides its own system and stream descriptor implementations, they should be used instead.
  * Otherwise, this {@link GenericOutputDescriptor} may be used to provide Samza-specific properties of the output stream.
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.descriptors;
+package org.apache.samza.system.descriptors;
 
 
-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;
 
 /**
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.descriptors.base.stream;
+package org.apache.samza.system.descriptors;
 
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
-import org.apache.samza.operators.descriptors.base.system.SystemDescriptor;
-import org.apache.samza.operators.functions.InputTransformer;
 import org.apache.samza.serializers.Serde;
 import org.apache.samza.system.SystemStreamMetadata.OffsetType;
 
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.functions;
+package org.apache.samza.system.descriptors;
 
 import java.io.Serializable;
+import org.apache.samza.operators.functions.ClosableFunction;
+import org.apache.samza.operators.functions.InitableFunction;
 import org.apache.samza.system.IncomingMessageEnvelope;
 
 /**
@@ -16,9 +16,8 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.descriptors.base.stream;
+package org.apache.samza.system.descriptors;
 
-import org.apache.samza.operators.descriptors.base.system.SystemDescriptor;
 import org.apache.samza.serializers.Serde;
 
 /**
@@ -16,9 +16,8 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.descriptors.base.system;
+package org.apache.samza.system.descriptors;
 
-import org.apache.samza.operators.descriptors.base.stream.OutputDescriptor;
 import org.apache.samza.serializers.Serde;
 
 
@@ -16,9 +16,8 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.descriptors.base.system;
+package org.apache.samza.system.descriptors;
 
-import org.apache.samza.operators.descriptors.base.stream.InputDescriptor;
 import org.apache.samza.serializers.Serde;
 
 
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.descriptors.base.stream;
+package org.apache.samza.system.descriptors;
 
 import com.google.common.base.Preconditions;
 
@@ -26,7 +26,6 @@ import java.util.Map;
 import java.util.Optional;
 import java.util.regex.Pattern;
 import org.apache.commons.lang3.StringUtils;
-import org.apache.samza.operators.descriptors.base.system.SystemDescriptor;
 import org.apache.samza.serializers.Serde;
 
 /**
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.functions;
+package org.apache.samza.system.descriptors;
 
 import java.io.Serializable;
-import org.apache.samza.application.StreamApplicationDescriptor;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 import org.apache.samza.operators.MessageStream;
-import org.apache.samza.operators.descriptors.base.stream.InputDescriptor;
 
 /**
  * Expands the provided {@link InputDescriptor} to a sub-DAG of one or more operators on the {@link StreamApplicationDescriptor},
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.descriptors.base.system;
+package org.apache.samza.system.descriptors;
 
 import com.google.common.base.Preconditions;
 
@@ -25,8 +25,6 @@ import java.util.Map;
 import java.util.Optional;
 import java.util.regex.Pattern;
 import org.apache.commons.lang3.StringUtils;
-import org.apache.samza.operators.functions.InputTransformer;
-import org.apache.samza.operators.functions.StreamExpander;
 import org.apache.samza.system.SystemStreamMetadata.OffsetType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -108,7 +106,7 @@ public abstract class SystemDescriptor<SubClass extends SystemDescriptor<SubClas
    *              and reprocess the entire available message history.
    * </ul>
    * This property is for all streams obtained using this system descriptor. To set it for an individual stream,
-   * see {@link org.apache.samza.operators.descriptors.base.stream.InputDescriptor#withOffsetDefault}.
+   * see {@link InputDescriptor#withOffsetDefault}.
    * If both are defined, the stream-level definition takes precedence.
    *
    * @param offsetType offset type to start processing from
@@ -16,9 +16,8 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.descriptors.base.system;
+package org.apache.samza.system.descriptors;
 
-import org.apache.samza.operators.descriptors.base.stream.InputDescriptor;
 import org.apache.samza.serializers.Serde;
 
 /**
index 7b8754a..296edf4 100644 (file)
@@ -22,7 +22,7 @@ import java.util.List;
 
 import org.apache.samza.annotation.InterfaceStability;
 import org.apache.samza.config.Config;
-import org.apache.samza.operators.TableDescriptor;
+import org.apache.samza.table.descriptors.TableDescriptor;
 
 
 /**
index 82577c4..2883a93 100644 (file)
@@ -32,8 +32,9 @@ import org.apache.samza.storage.SideInputsProcessor;
 /**
  * TableSpec is a blueprint for creating, validating, or simply describing a table in the runtime environment.
  *
- * It is typically created indirectly by constructing an instance of {@link org.apache.samza.operators.TableDescriptor},
- * and then invoke <code>BaseTableDescriptor.getTableSpec()</code>.
+ * It is typically created indirectly by constructing an instance of
+ * {@link org.apache.samza.table.descriptors.TableDescriptor}, and then invoke
+ * <code>BaseTableDescriptor.getTableSpec()</code>.
  *
  * It has specific attributes for common behaviors that Samza uses.
  *
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators;
+package org.apache.samza.table.descriptors;
 
 import org.apache.samza.annotation.InterfaceStability;
 
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.table;
+package org.apache.samza.table.descriptors;
 
 import java.util.Map;
 import org.apache.samza.annotation.InterfaceStability;
 import org.apache.samza.config.Config;
 import org.apache.samza.context.Context;
+import org.apache.samza.table.Table;
 
 /**
  * A table provider provides the implementation for a table. It ensures a table is
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.table;
+package org.apache.samza.table.descriptors;
 
 import org.apache.samza.annotation.InterfaceStability;
+import org.apache.samza.table.TableSpec;
 
 
 /**
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.descriptors;
+package org.apache.samza.system.descriptors;
 
 import java.util.Collections;
-import org.apache.samza.operators.descriptors.expanding.ExampleExpandingInputDescriptor;
-import org.apache.samza.operators.descriptors.expanding.ExampleExpandingOutputDescriptor;
-import org.apache.samza.operators.descriptors.expanding.ExampleExpandingSystemDescriptor;
+import org.apache.samza.system.descriptors.examples.expanding.ExampleExpandingInputDescriptor;
+import org.apache.samza.system.descriptors.examples.expanding.ExampleExpandingOutputDescriptor;
+import org.apache.samza.system.descriptors.examples.expanding.ExampleExpandingSystemDescriptor;
 import org.apache.samza.serializers.IntegerSerde;
 import org.apache.samza.system.SystemStreamMetadata;
 import org.junit.Test;
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.descriptors;
+package org.apache.samza.system.descriptors;
 
 import java.util.Collections;
-import org.apache.samza.operators.descriptors.serde.ExampleSimpleInputDescriptor;
-import org.apache.samza.operators.descriptors.serde.ExampleSimpleOutputDescriptor;
-import org.apache.samza.operators.descriptors.serde.ExampleSimpleSystemDescriptor;
+import org.apache.samza.system.descriptors.examples.serde.ExampleSimpleInputDescriptor;
+import org.apache.samza.system.descriptors.examples.serde.ExampleSimpleOutputDescriptor;
+import org.apache.samza.system.descriptors.examples.serde.ExampleSimpleSystemDescriptor;
 import org.apache.samza.serializers.IntegerSerde;
 import org.apache.samza.system.SystemStreamMetadata;
 import org.junit.Test;
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.descriptors;
+package org.apache.samza.system.descriptors;
 
 import java.util.Collections;
-import org.apache.samza.operators.descriptors.transforming.ExampleTransformingInputDescriptor;
-import org.apache.samza.operators.descriptors.transforming.ExampleTransformingOutputDescriptor;
-import org.apache.samza.operators.descriptors.transforming.ExampleTransformingSystemDescriptor;
+import org.apache.samza.system.descriptors.examples.transforming.ExampleTransformingInputDescriptor;
+import org.apache.samza.system.descriptors.examples.transforming.ExampleTransformingOutputDescriptor;
+import org.apache.samza.system.descriptors.examples.transforming.ExampleTransformingSystemDescriptor;
 import org.apache.samza.serializers.IntegerSerde;
 import org.apache.samza.system.SystemStreamMetadata;
 import org.junit.Test;
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.descriptors.expanding;
+package org.apache.samza.system.descriptors.examples.expanding;
 
-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.system.descriptors.InputDescriptor;
+import org.apache.samza.system.descriptors.SystemDescriptor;
+import org.apache.samza.system.descriptors.InputTransformer;
 import org.apache.samza.serializers.Serde;
 
 public class ExampleExpandingInputDescriptor<StreamMessageType> extends InputDescriptor<StreamMessageType, ExampleExpandingInputDescriptor<StreamMessageType>> {
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.descriptors.expanding;
+package org.apache.samza.system.descriptors.examples.expanding;
 
-import org.apache.samza.operators.descriptors.base.stream.OutputDescriptor;
-import org.apache.samza.operators.descriptors.base.system.SystemDescriptor;
+import org.apache.samza.system.descriptors.OutputDescriptor;
+import org.apache.samza.system.descriptors.SystemDescriptor;
 import org.apache.samza.serializers.Serde;
 
 public class ExampleExpandingOutputDescriptor<StreamMessageType> extends OutputDescriptor<StreamMessageType, ExampleExpandingOutputDescriptor<StreamMessageType>> {
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.descriptors.expanding;
+package org.apache.samza.system.descriptors.examples.expanding;
 
+import org.apache.samza.system.descriptors.ExpandingInputDescriptorProvider;
+import org.apache.samza.system.descriptors.OutputDescriptorProvider;
+import org.apache.samza.system.descriptors.SystemDescriptor;
 import org.apache.samza.operators.MessageStream;
-import org.apache.samza.operators.descriptors.base.system.ExpandingInputDescriptorProvider;
-import org.apache.samza.operators.descriptors.base.system.OutputDescriptorProvider;
-import org.apache.samza.operators.descriptors.base.system.SystemDescriptor;
-import org.apache.samza.operators.functions.InputTransformer;
+import org.apache.samza.system.descriptors.InputTransformer;
 import org.apache.samza.serializers.Serde;
 import org.apache.samza.system.IncomingMessageEnvelope;
 
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.descriptors.serde;
+package org.apache.samza.system.descriptors.examples.serde;
 
-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.system.descriptors.InputDescriptor;
+import org.apache.samza.system.descriptors.SystemDescriptor;
+import org.apache.samza.system.descriptors.InputTransformer;
 import org.apache.samza.serializers.Serde;
 
 public class ExampleSimpleInputDescriptor<StreamMessageType> extends InputDescriptor<StreamMessageType, ExampleSimpleInputDescriptor<StreamMessageType>> {
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.descriptors.serde;
+package org.apache.samza.system.descriptors.examples.serde;
 
-import org.apache.samza.operators.descriptors.base.stream.OutputDescriptor;
-import org.apache.samza.operators.descriptors.base.system.SystemDescriptor;
+import org.apache.samza.system.descriptors.OutputDescriptor;
+import org.apache.samza.system.descriptors.SystemDescriptor;
 import org.apache.samza.serializers.Serde;
 
 public class ExampleSimpleOutputDescriptor<StreamMessageType> extends OutputDescriptor<StreamMessageType, ExampleSimpleOutputDescriptor<StreamMessageType>> {
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.descriptors.serde;
+package org.apache.samza.system.descriptors.examples.serde;
 
-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.system.descriptors.OutputDescriptorProvider;
+import org.apache.samza.system.descriptors.SimpleInputDescriptorProvider;
+import org.apache.samza.system.descriptors.SystemDescriptor;
 import org.apache.samza.serializers.Serde;
 
 public class ExampleSimpleSystemDescriptor extends SystemDescriptor<ExampleSimpleSystemDescriptor>
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.descriptors.transforming;
+package org.apache.samza.system.descriptors.examples.transforming;
 
-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.system.descriptors.InputDescriptor;
+import org.apache.samza.system.descriptors.SystemDescriptor;
+import org.apache.samza.system.descriptors.InputTransformer;
 import org.apache.samza.serializers.Serde;
 
 public class ExampleTransformingInputDescriptor<StreamMessageType> extends InputDescriptor<StreamMessageType, ExampleTransformingInputDescriptor<StreamMessageType>> {
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.descriptors.transforming;
+package org.apache.samza.system.descriptors.examples.transforming;
 
-import org.apache.samza.operators.descriptors.base.stream.OutputDescriptor;
-import org.apache.samza.operators.descriptors.base.system.SystemDescriptor;
+import org.apache.samza.system.descriptors.OutputDescriptor;
+import org.apache.samza.system.descriptors.SystemDescriptor;
 import org.apache.samza.serializers.Serde;
 
 public class ExampleTransformingOutputDescriptor<StreamMessageType> extends OutputDescriptor<StreamMessageType, ExampleTransformingOutputDescriptor<StreamMessageType>> {
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.descriptors.transforming;
+package org.apache.samza.system.descriptors.examples.transforming;
 
-import org.apache.samza.operators.descriptors.base.system.OutputDescriptorProvider;
-import org.apache.samza.operators.descriptors.base.system.SystemDescriptor;
-import org.apache.samza.operators.descriptors.base.system.TransformingInputDescriptorProvider;
+import org.apache.samza.system.descriptors.OutputDescriptorProvider;
+import org.apache.samza.system.descriptors.SystemDescriptor;
+import org.apache.samza.system.descriptors.TransformingInputDescriptorProvider;
 import org.apache.samza.serializers.Serde;
 
 public class ExampleTransformingSystemDescriptor extends SystemDescriptor<ExampleTransformingSystemDescriptor>
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.system.eventhub;
+package org.apache.samza.system.eventhub.descriptors;
 
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.samza.config.ConfigException;
-import org.apache.samza.operators.descriptors.base.stream.InputDescriptor;
-import org.apache.samza.operators.descriptors.base.system.SystemDescriptor;
+import org.apache.samza.system.descriptors.InputDescriptor;
+import org.apache.samza.system.descriptors.SystemDescriptor;
 import org.apache.samza.serializers.Serde;
+import org.apache.samza.system.eventhub.EventHubConfig;
 
 
 /**
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.system.eventhub;
+package org.apache.samza.system.eventhub.descriptors;
 
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.samza.config.ConfigException;
-import org.apache.samza.operators.descriptors.base.stream.OutputDescriptor;
-import org.apache.samza.operators.descriptors.base.system.SystemDescriptor;
+import org.apache.samza.system.descriptors.OutputDescriptor;
+import org.apache.samza.system.descriptors.SystemDescriptor;
 import org.apache.samza.serializers.Serde;
+import org.apache.samza.system.eventhub.EventHubConfig;
 
 /**
  * A descriptor for an Event Hubs output stream
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.system.eventhub;
+package org.apache.samza.system.eventhub.descriptors;
 
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import org.apache.samza.operators.descriptors.base.system.SystemDescriptor;
+import org.apache.samza.system.descriptors.SystemDescriptor;
 import org.apache.samza.serializers.Serde;
+import org.apache.samza.system.eventhub.EventHubConfig;
+import org.apache.samza.system.eventhub.EventHubSystemFactory;
 import org.apache.samza.system.eventhub.producer.EventHubSystemProducer.PartitioningMethod;
 
 
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.system.eventhub;
+package org.apache.samza.system.eventhub.descriptors;
 
 import java.util.Map;
 import org.apache.samza.config.ConfigException;
@@ -24,6 +24,7 @@ import org.apache.samza.operators.KV;
 import org.apache.samza.serializers.IntegerSerde;
 import org.apache.samza.serializers.KVSerde;
 import org.apache.samza.serializers.StringSerde;
+import org.apache.samza.system.eventhub.EventHubConfig;
 import org.junit.Test;
 
 import static org.junit.Assert.assertEquals;
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.system.eventhub;
+package org.apache.samza.system.eventhub.descriptors;
 
 import java.util.Map;
 import org.apache.samza.config.ConfigException;
@@ -24,6 +24,7 @@ import org.apache.samza.operators.KV;
 import org.apache.samza.serializers.IntegerSerde;
 import org.apache.samza.serializers.KVSerde;
 import org.apache.samza.serializers.StringSerde;
+import org.apache.samza.system.eventhub.EventHubConfig;
 import org.junit.Test;
 
 import static org.junit.Assert.assertEquals;
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.system.eventhub;
+package org.apache.samza.system.eventhub.descriptors;
 
 import java.util.Map;
 import org.apache.samza.serializers.IntegerSerde;
 import org.apache.samza.serializers.KVSerde;
 import org.apache.samza.serializers.StringSerde;
+import org.apache.samza.system.eventhub.EventHubConfig;
 import org.apache.samza.system.eventhub.producer.EventHubSystemProducer.PartitioningMethod;
 import org.junit.Test;
 
index e9e2635..2b29a2b 100644 (file)
@@ -18,6 +18,7 @@
  */
 package org.apache.samza.application;
 
+import org.apache.samza.application.descriptors.TaskApplicationDescriptor;
 import org.apache.samza.task.TaskFactoryUtil;
 
 /**
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.application;
+package org.apache.samza.application.descriptors;
 
 import java.util.Collections;
 import java.util.HashMap;
@@ -24,17 +24,18 @@ import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import org.apache.samza.application.SamzaApplication;
 import org.apache.samza.config.Config;
 import org.apache.samza.context.ApplicationContainerContext;
 import org.apache.samza.context.ApplicationContainerContextFactory;
 import org.apache.samza.context.ApplicationTaskContext;
 import org.apache.samza.context.ApplicationTaskContextFactory;
+import org.apache.samza.system.descriptors.InputDescriptor;
+import org.apache.samza.system.descriptors.OutputDescriptor;
 import org.apache.samza.metrics.MetricsReporterFactory;
 import org.apache.samza.operators.KV;
-import org.apache.samza.operators.TableDescriptor;
-import org.apache.samza.operators.descriptors.base.stream.InputDescriptor;
-import org.apache.samza.operators.descriptors.base.stream.OutputDescriptor;
-import org.apache.samza.operators.descriptors.base.system.SystemDescriptor;
+import org.apache.samza.table.descriptors.TableDescriptor;
+import org.apache.samza.system.descriptors.SystemDescriptor;
 import org.apache.samza.operators.spec.InputOperatorSpec;
 import org.apache.samza.runtime.ProcessorLifecycleListener;
 import org.apache.samza.runtime.ProcessorLifecycleListenerFactory;
@@ -192,7 +193,8 @@ public abstract class ApplicationDescriptorImpl<S extends ApplicationDescriptor>
   /**
    * Get the map of all {@link InputOperatorSpec}s in this applicaiton
    *
-   * @return an immutable map from streamId to {@link InputOperatorSpec}. Default to empty map for low-level {@link TaskApplication}
+   * @return an immutable map from streamId to {@link InputOperatorSpec}. Default to empty map for low-level
+   * {@link org.apache.samza.application.TaskApplication}
    */
   public Map<String, InputOperatorSpec> getInputOperators() {
     return Collections.EMPTY_MAP;
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.application;
+package org.apache.samza.application.descriptors;
 
+import org.apache.samza.application.SamzaApplication;
+import org.apache.samza.application.StreamApplication;
+import org.apache.samza.application.TaskApplication;
 import org.apache.samza.config.Config;
 
 
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.application;
+package org.apache.samza.application.descriptors;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
@@ -30,21 +30,22 @@ import java.util.Set;
 import java.util.regex.Pattern;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.samza.SamzaException;
+import org.apache.samza.application.StreamApplication;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.JobConfig;
-import org.apache.samza.operators.BaseTableDescriptor;
+import org.apache.samza.system.descriptors.InputDescriptor;
+import org.apache.samza.system.descriptors.OutputDescriptor;
+import org.apache.samza.table.descriptors.BaseTableDescriptor;
 import org.apache.samza.operators.KV;
 import org.apache.samza.operators.MessageStream;
 import org.apache.samza.operators.MessageStreamImpl;
 import org.apache.samza.operators.OperatorSpecGraph;
 import org.apache.samza.operators.OutputStream;
-import org.apache.samza.operators.TableDescriptor;
+import org.apache.samza.table.descriptors.TableDescriptor;
 import org.apache.samza.operators.TableImpl;
-import org.apache.samza.operators.descriptors.base.stream.InputDescriptor;
-import org.apache.samza.operators.descriptors.base.stream.OutputDescriptor;
-import org.apache.samza.operators.descriptors.base.system.SystemDescriptor;
-import org.apache.samza.operators.functions.InputTransformer;
-import org.apache.samza.operators.functions.StreamExpander;
+import org.apache.samza.system.descriptors.SystemDescriptor;
+import org.apache.samza.system.descriptors.InputTransformer;
+import org.apache.samza.system.descriptors.StreamExpander;
 import org.apache.samza.operators.spec.InputOperatorSpec;
 import org.apache.samza.operators.spec.OperatorSpec.OpCode;
 import org.apache.samza.operators.spec.OperatorSpecs;
@@ -54,7 +55,7 @@ import org.apache.samza.serializers.KVSerde;
 import org.apache.samza.serializers.Serde;
 import org.apache.samza.table.Table;
 import org.apache.samza.table.TableSpec;
-import org.apache.samza.table.hybrid.BaseHybridTableDescriptor;
+import org.apache.samza.table.descriptors.BaseHybridTableDescriptor;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.application;
+package org.apache.samza.application.descriptors;
 
 import com.google.common.base.Preconditions;
 import java.util.Collections;
@@ -24,12 +24,13 @@ import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Set;
+import org.apache.samza.application.TaskApplication;
 import org.apache.samza.config.Config;
-import org.apache.samza.operators.BaseTableDescriptor;
-import org.apache.samza.operators.TableDescriptor;
-import org.apache.samza.operators.descriptors.base.stream.InputDescriptor;
-import org.apache.samza.operators.descriptors.base.stream.OutputDescriptor;
-import org.apache.samza.operators.descriptors.base.system.SystemDescriptor;
+import org.apache.samza.system.descriptors.InputDescriptor;
+import org.apache.samza.system.descriptors.OutputDescriptor;
+import org.apache.samza.table.descriptors.BaseTableDescriptor;
+import org.apache.samza.table.descriptors.TableDescriptor;
+import org.apache.samza.system.descriptors.SystemDescriptor;
 import org.apache.samza.task.TaskFactory;
 
 
index ed013c4..06f5606 100644 (file)
@@ -59,9 +59,9 @@ public class JavaTableConfig extends MapConfig {
   }
 
   /**
-   * Get the {@link org.apache.samza.table.TableProviderFactory} class for a table
+   * Get the {@link org.apache.samza.table.descriptors.TableProviderFactory} class for a table
    * @param tableId Id of the table
-   * @return the {@link org.apache.samza.table.TableProviderFactory} class name
+   * @return the {@link org.apache.samza.table.descriptors.TableProviderFactory} class name
    */
   public String getTableProviderFactory(String tableId) {
     return get(String.format(TABLE_PROVIDER_FACTORY, tableId), null);
index b80f7df..10a4215 100644 (file)
@@ -35,15 +35,15 @@ import java.util.Set;
 import java.util.stream.Collectors;
 import org.apache.commons.collections4.ListUtils;
 import org.apache.samza.SamzaException;
-import org.apache.samza.application.ApplicationDescriptor;
-import org.apache.samza.application.ApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.ApplicationDescriptor;
+import org.apache.samza.application.descriptors.ApplicationDescriptorImpl;
 import org.apache.samza.application.LegacyTaskApplication;
 import org.apache.samza.config.ApplicationConfig;
 import org.apache.samza.config.ClusterManagerConfig;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.JobConfig;
 import org.apache.samza.config.StreamConfig;
-import org.apache.samza.operators.BaseTableDescriptor;
+import org.apache.samza.table.descriptors.BaseTableDescriptor;
 import org.apache.samza.operators.spec.InputOperatorSpec;
 import org.apache.samza.operators.spec.OperatorSpec;
 import org.apache.samza.operators.spec.StreamTableJoinOperatorSpec;
index f43b24e..6d9faf3 100644 (file)
@@ -30,8 +30,8 @@ import java.util.Map;
 import java.util.Queue;
 import java.util.Set;
 import java.util.stream.Collectors;
-import org.apache.samza.application.ApplicationDescriptor;
-import org.apache.samza.application.ApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.ApplicationDescriptor;
+import org.apache.samza.application.descriptors.ApplicationDescriptorImpl;
 import org.apache.samza.config.ApplicationConfig;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.JobConfig;
index 18705e4..4a2a235 100644 (file)
@@ -200,8 +200,8 @@ import org.codehaus.jackson.map.ObjectMapper;
   }
 
   /**
-   * Create JSON POJO for a {@link JobNode}, including the {@link org.apache.samza.application.ApplicationDescriptorImpl}
-   * for this job
+   * Create JSON POJO for a {@link JobNode}, including the
+   * {@link org.apache.samza.application.descriptors.ApplicationDescriptorImpl} for this job
    *
    * @param jobNode job node in the {@link JobGraph}
    * @return {@link org.apache.samza.execution.JobGraphJsonGenerator.JobNodeJson}
index af556f5..82b4178 100644 (file)
@@ -26,8 +26,8 @@ import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.stream.Collectors;
-import org.apache.samza.application.ApplicationDescriptor;
-import org.apache.samza.application.ApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.ApplicationDescriptor;
+import org.apache.samza.application.descriptors.ApplicationDescriptorImpl;
 import org.apache.samza.application.LegacyTaskApplication;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.JobConfig;
index 83f3f61..dc0fc59 100644 (file)
@@ -26,8 +26,8 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import org.apache.commons.lang3.StringUtils;
-import org.apache.samza.application.ApplicationDescriptor;
-import org.apache.samza.application.ApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.ApplicationDescriptor;
+import org.apache.samza.application.descriptors.ApplicationDescriptorImpl;
 import org.apache.samza.application.LegacyTaskApplication;
 import org.apache.samza.config.ApplicationConfig;
 import org.apache.samza.config.Config;
index 86aca0f..6ca5f3d 100644 (file)
@@ -23,8 +23,8 @@ import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import org.apache.samza.SamzaException;
-import org.apache.samza.application.ApplicationDescriptor;
-import org.apache.samza.application.ApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.ApplicationDescriptor;
+import org.apache.samza.application.descriptors.ApplicationDescriptorImpl;
 import org.apache.samza.config.ApplicationConfig;
 import org.apache.samza.config.JobConfig;
 import org.apache.samza.config.JobCoordinatorConfig;
index 54f86d5..13b29df 100644 (file)
@@ -21,8 +21,8 @@ package org.apache.samza.execution;
 import java.util.List;
 import java.util.UUID;
 import org.apache.samza.SamzaException;
-import org.apache.samza.application.ApplicationDescriptor;
-import org.apache.samza.application.ApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.ApplicationDescriptor;
+import org.apache.samza.application.descriptors.ApplicationDescriptorImpl;
 import org.apache.samza.config.ApplicationConfig;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.JobConfig;
index 09e4868..0f43c5e 100644 (file)
@@ -24,7 +24,7 @@ import java.time.Duration;
 import java.util.Collection;
 
 import org.apache.samza.SamzaException;
-import org.apache.samza.application.StreamApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptorImpl;
 import org.apache.samza.operators.functions.FilterFunction;
 import org.apache.samza.operators.functions.FlatMapFunction;
 import org.apache.samza.operators.functions.JoinFunction;
index 5329fd7..a83739d 100644 (file)
@@ -25,7 +25,7 @@ import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
-import org.apache.samza.application.StreamApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptorImpl;
 import org.apache.samza.operators.spec.InputOperatorSpec;
 import org.apache.samza.operators.spec.OperatorSpec;
 import org.apache.samza.operators.spec.OutputStreamImpl;
index 2a73064..fbeda3e 100644 (file)
@@ -20,7 +20,7 @@ package org.apache.samza.operators.impl;
 
 import org.apache.samza.context.Context;
 import org.apache.samza.operators.KV;
-import org.apache.samza.operators.functions.InputTransformer;
+import org.apache.samza.system.descriptors.InputTransformer;
 import org.apache.samza.operators.spec.InputOperatorSpec;
 import org.apache.samza.operators.spec.OperatorSpec;
 import org.apache.samza.system.IncomingMessageEnvelope;
index 1af4806..b467d60 100644 (file)
@@ -18,7 +18,7 @@
  */
 package org.apache.samza.operators.spec;
 
-import org.apache.samza.operators.functions.InputTransformer;
+import org.apache.samza.system.descriptors.InputTransformer;
 import org.apache.samza.operators.functions.ScheduledFunction;
 import org.apache.samza.operators.functions.WatermarkFunction;
 import org.apache.samza.serializers.Serde;
index 0442f7c..1886d1b 100644 (file)
@@ -103,7 +103,8 @@ public abstract class OperatorSpec<M, OM> implements Serializable {
   }
 
   /**
-   * Get the unique ID of this operator in the {@link org.apache.samza.application.StreamApplicationDescriptorImpl}.
+   * Get the unique ID of this operator in the
+   * {@link org.apache.samza.application.descriptors.StreamApplicationDescriptorImpl}.
    * @return  the unique operator ID
    */
   public final String getOpId() {
index 6ebbdae..8d3ff60 100644 (file)
@@ -22,7 +22,7 @@ package org.apache.samza.operators.spec;
 import org.apache.samza.operators.KV;
 import org.apache.samza.operators.functions.FilterFunction;
 import org.apache.samza.operators.functions.FlatMapFunction;
-import org.apache.samza.operators.functions.InputTransformer;
+import org.apache.samza.system.descriptors.InputTransformer;
 import org.apache.samza.operators.functions.JoinFunction;
 import org.apache.samza.operators.functions.MapFunction;
 import org.apache.samza.operators.functions.SinkFunction;
index 4db8e60..44f62d9 100644 (file)
@@ -18,7 +18,7 @@
  */
 package org.apache.samza.operators.stream;
 
-import org.apache.samza.application.StreamApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptorImpl;
 import org.apache.samza.operators.MessageStreamImpl;
 import org.apache.samza.operators.OutputStream;
 import org.apache.samza.operators.spec.InputOperatorSpec;
index a5eeba1..85306ef 100644 (file)
@@ -32,10 +32,10 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import org.apache.samza.SamzaException;
-import org.apache.samza.application.ApplicationDescriptor;
-import org.apache.samza.application.ApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.ApplicationDescriptor;
+import org.apache.samza.application.descriptors.ApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.ApplicationDescriptorUtil;
 import org.apache.samza.application.SamzaApplication;
-import org.apache.samza.application.ApplicationDescriptorUtil;
 import org.apache.samza.config.ApplicationConfig;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.JobConfig;
index 94ff1eb..c5c0d78 100644 (file)
@@ -24,9 +24,9 @@ import java.util.Map;
 import java.util.Random;
 import org.slf4j.MDC;
 import org.apache.samza.SamzaException;
-import org.apache.samza.application.ApplicationDescriptor;
-import org.apache.samza.application.ApplicationDescriptorImpl;
-import org.apache.samza.application.ApplicationDescriptorUtil;
+import org.apache.samza.application.descriptors.ApplicationDescriptor;
+import org.apache.samza.application.descriptors.ApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.ApplicationDescriptorUtil;
 import org.apache.samza.application.ApplicationUtil;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.JobConfig;
index 69eb5fe..1b38c9b 100644 (file)
@@ -22,10 +22,10 @@ package org.apache.samza.runtime;
 import java.time.Duration;
 import java.util.List;
 import org.apache.samza.SamzaException;
-import org.apache.samza.application.ApplicationDescriptor;
-import org.apache.samza.application.ApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.ApplicationDescriptor;
+import org.apache.samza.application.descriptors.ApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.ApplicationDescriptorUtil;
 import org.apache.samza.application.SamzaApplication;
-import org.apache.samza.application.ApplicationDescriptorUtil;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.JobConfig;
 import org.apache.samza.execution.RemoteJobPlanner;
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.operators.descriptors;
+package org.apache.samza.system.descriptors;
 
 
 import com.google.common.annotations.VisibleForTesting;
 
-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;
 
 /**
index 03be758..05454a6 100644 (file)
@@ -27,8 +27,10 @@ import java.util.Map;
 
 import org.apache.samza.config.Config;
 import org.apache.samza.config.JavaTableConfig;
-import org.apache.samza.operators.BaseTableDescriptor;
-import org.apache.samza.operators.TableDescriptor;
+import org.apache.samza.table.descriptors.TableProvider;
+import org.apache.samza.table.descriptors.TableProviderFactory;
+import org.apache.samza.table.descriptors.BaseTableDescriptor;
+import org.apache.samza.table.descriptors.TableDescriptor;
 import org.apache.samza.operators.TableImpl;
 import org.apache.samza.util.Util;
 import org.slf4j.Logger;
index d7b15a4..1d1e8b1 100644 (file)
@@ -23,6 +23,8 @@ import org.apache.samza.SamzaException;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.JavaTableConfig;
 import org.apache.samza.context.Context;
+import org.apache.samza.table.descriptors.TableProvider;
+import org.apache.samza.table.descriptors.TableProviderFactory;
 import org.apache.samza.serializers.KVSerde;
 import org.apache.samza.serializers.Serde;
 import org.apache.samza.util.Util;
@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.apache.samza.table.caching;
+package org.apache.samza.table.caching.descriptors;
 
 import java.time.Duration;
 import java.util.Arrays;
@@ -25,16 +25,17 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
-import org.apache.samza.operators.BaseTableDescriptor;
-import org.apache.samza.operators.TableDescriptor;
+import org.apache.samza.table.descriptors.BaseTableDescriptor;
+import org.apache.samza.table.descriptors.TableDescriptor;
 import org.apache.samza.table.TableSpec;
-import org.apache.samza.table.hybrid.BaseHybridTableDescriptor;
+import org.apache.samza.table.descriptors.BaseHybridTableDescriptor;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 
 
 /**
- * Table descriptor for {@link CachingTable}.
+ * Table descriptor for {@link org.apache.samza.table.caching.CachingTable}.
  * @param <K> type of the key in the cache
  * @param <V> type of the value in the cache
  */
@@ -60,7 +61,7 @@ public class CachingTableDescriptor<K, V> extends BaseHybridTableDescriptor<K, V
   /**
    * Constructs a table descriptor instance and specify a cache (as Table descriptor)
    * to be used for caching. Cache get is not synchronized with put for better parallelism
-   * in the read path of {@link CachingTable}. As such, cache table implementation is
+   * in the read path of {@link org.apache.samza.table.caching.CachingTable}. As such, cache table implementation is
    * expected to be thread-safe for concurrent accesses.
    *
    * @param tableId Id of the table, it must conform to pattern { @literal [\\d\\w-_]+ }
@@ -151,7 +152,8 @@ public class CachingTableDescriptor<K, V> extends BaseHybridTableDescriptor<K, V
   }
 
   @Override
-  protected void validate() {
+  @VisibleForTesting
+  public void validate() {
     super.validate();
     Preconditions.checkNotNull(table, "Actual table is required.");
     if (cache == null) {
@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.apache.samza.table.caching;
+package org.apache.samza.table.caching.descriptors;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -27,8 +27,9 @@ import org.apache.samza.table.ReadWriteTable;
 import org.apache.samza.table.ReadableTable;
 import org.apache.samza.table.Table;
 import org.apache.samza.table.TableSpec;
+import org.apache.samza.table.caching.CachingTable;
 import org.apache.samza.table.caching.guava.GuavaCacheTable;
-import org.apache.samza.table.utils.BaseTableProvider;
+import org.apache.samza.table.utils.descriptors.BaseTableProvider;
 
 import com.google.common.cache.CacheBuilder;
 
  * under the License.
  */
 
-package org.apache.samza.table.caching;
+package org.apache.samza.table.caching.descriptors;
 
-import org.apache.samza.table.TableProvider;
-import org.apache.samza.table.TableProviderFactory;
+import org.apache.samza.table.descriptors.TableProvider;
+import org.apache.samza.table.descriptors.TableProviderFactory;
 import org.apache.samza.table.TableSpec;
 
 /**
- * Table provider factory for {@link CachingTable}.
+ * Table provider factory for {@link org.apache.samza.table.caching.CachingTable}.
  */
 public class CachingTableProviderFactory implements TableProviderFactory {
   @Override
  * under the License.
  */
 
-package org.apache.samza.table.caching.guava;
+package org.apache.samza.table.caching.guava.descriptors;
 
 import java.util.HashMap;
 import java.util.Map;
 
-import org.apache.samza.operators.BaseTableDescriptor;
+import org.apache.samza.table.descriptors.BaseTableDescriptor;
 import org.apache.samza.table.TableSpec;
 import org.apache.samza.table.utils.SerdeUtils;
 
@@ -31,7 +31,7 @@ import com.google.common.cache.Cache;
 
 
 /**
- * Table descriptor for {@link GuavaCacheTable}.
+ * Table descriptor for {@link org.apache.samza.table.caching.guava.GuavaCacheTable}.
  * @param <K> type of the key in the cache
  * @param <V> type of the value in the cache
  */
  * under the License.
  */
 
-package org.apache.samza.table.caching.guava;
+package org.apache.samza.table.caching.guava.descriptors;
 
 import java.util.ArrayList;
 import java.util.List;
 
 import org.apache.samza.table.Table;
 import org.apache.samza.table.TableSpec;
-import org.apache.samza.table.utils.BaseTableProvider;
+import org.apache.samza.table.caching.guava.GuavaCacheTable;
+import org.apache.samza.table.utils.descriptors.BaseTableProvider;
 import org.apache.samza.table.utils.SerdeUtils;
 
 import com.google.common.cache.Cache;
  * under the License.
  */
 
-package org.apache.samza.table.caching.guava;
+package org.apache.samza.table.caching.guava.descriptors;
 
-import org.apache.samza.table.TableProvider;
-import org.apache.samza.table.TableProviderFactory;
+import org.apache.samza.table.descriptors.TableProvider;
+import org.apache.samza.table.descriptors.TableProviderFactory;
 import org.apache.samza.table.TableSpec;
 
 /**
- * Table provider factory for {@link GuavaCacheTable}.
+ * Table provider factory for {@link org.apache.samza.table.caching.guava.GuavaCacheTable}.
  */
 public class GuavaCacheTableProviderFactory implements TableProviderFactory {
   @Override
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.table.hybrid;
+package org.apache.samza.table.descriptors;
 
 import java.util.List;
-import org.apache.samza.operators.BaseTableDescriptor;
-import org.apache.samza.operators.TableDescriptor;
 
 /**
  * Base class for hybrid table descriptors. A hybrid table consists of one or more
index 4cbc270..52bdc71 100644 (file)
@@ -44,9 +44,8 @@ public class RemoteReadWriteTable<K, V> extends RemoteReadableTable<K, V> implem
 
   private DefaultTableWriteMetrics writeMetrics;
 
-  @VisibleForTesting
-  final TableWriteFunction<K, V> writeFn;
-  final TableRateLimiter writeRateLimiter;
+  protected final TableWriteFunction<K, V> writeFn;
+  protected final TableRateLimiter writeRateLimiter;
 
   public RemoteReadWriteTable(String tableId, TableReadFunction readFn, TableWriteFunction writeFn,
       TableRateLimiter<K, V> readRateLimiter, TableRateLimiter<K, V> writeRateLimiter,
@@ -219,4 +218,14 @@ public class RemoteReadWriteTable<K, V> extends RemoteReadableTable<K, V> implem
     writeFn.close();
     super.close();
   }
+
+  @VisibleForTesting
+  public TableWriteFunction<K, V> getWriteFn() {
+    return writeFn;
+  }
+
+  @VisibleForTesting
+  public TableRateLimiter getWriteRateLimiter() {
+    return writeRateLimiter;
+  }
 }
index 9487e39..0ae2728 100644 (file)
@@ -77,13 +77,11 @@ public class RemoteReadableTable<K, V> implements ReadableTable<K, V> {
 
   protected final ExecutorService callbackExecutor;
   protected final ExecutorService tableExecutor;
+  protected final TableReadFunction<K, V> readFn;
+  protected final TableRateLimiter<K, V> readRateLimiter;
 
   private DefaultTableReadMetrics readMetrics;
 
-  @VisibleForTesting
-  final TableReadFunction<K, V> readFn;
-  final TableRateLimiter<K, V> readRateLimiter;
-
   /**
    * Construct a RemoteReadableTable instance
    * @param tableId table id
@@ -298,4 +296,24 @@ public class RemoteReadableTable<K, V> implements ReadableTable<K, V> {
   public void close() {
     readFn.close();
   }
+
+  @VisibleForTesting
+  public ExecutorService getCallbackExecutor() {
+    return callbackExecutor;
+  }
+
+  @VisibleForTesting
+  public ExecutorService getTableExecutor() {
+    return tableExecutor;
+  }
+
+  @VisibleForTesting
+  public TableReadFunction<K, V> getReadFn() {
+    return readFn;
+  }
+
+  @VisibleForTesting
+  public TableRateLimiter<K, V> getReadRateLimiter() {
+    return readRateLimiter;
+  }
 }
  * under the License.
  */
 
-package org.apache.samza.table.remote;
+package org.apache.samza.table.remote.descriptors;
 
 import java.util.HashMap;
 import java.util.Map;
 
-import org.apache.samza.operators.BaseTableDescriptor;
+import org.apache.samza.table.descriptors.BaseTableDescriptor;
 import org.apache.samza.serializers.KVSerde;
 import org.apache.samza.table.TableSpec;
+import org.apache.samza.table.remote.TableRateLimiter;
+import org.apache.samza.table.remote.TableReadFunction;
+import org.apache.samza.table.remote.TableWriteFunction;
 import org.apache.samza.table.retry.TableRetryPolicy;
 import org.apache.samza.table.utils.SerdeUtils;
 import org.apache.samza.util.EmbeddedTaggedRateLimiter;
  * under the License.
  */
 
-package org.apache.samza.table.remote;
+package org.apache.samza.table.remote.descriptors;
 
 import org.apache.samza.table.Table;
 import org.apache.samza.table.TableSpec;
+import org.apache.samza.table.remote.RemoteReadWriteTable;
+import org.apache.samza.table.remote.RemoteReadableTable;
+import org.apache.samza.table.remote.TableRateLimiter;
+import org.apache.samza.table.remote.TableReadFunction;
+import org.apache.samza.table.remote.TableWriteFunction;
 import org.apache.samza.table.retry.RetriableReadFunction;
 import org.apache.samza.table.retry.RetriableWriteFunction;
 import org.apache.samza.table.retry.TableRetryPolicy;
-import org.apache.samza.table.utils.BaseTableProvider;
+import org.apache.samza.table.utils.descriptors.BaseTableProvider;
 import org.apache.samza.table.utils.SerdeUtils;
 import org.apache.samza.table.utils.TableMetricsUtil;
 import org.apache.samza.util.RateLimiter;
@@ -37,9 +42,6 @@ import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 
-import static org.apache.samza.table.remote.RemoteTableDescriptor.RL_READ_TAG;
-import static org.apache.samza.table.remote.RemoteTableDescriptor.RL_WRITE_TAG;
-
 
 /**
  * Provide for remote table instances
@@ -86,7 +88,7 @@ public class RemoteTableProvider extends BaseTableProvider {
       rateLimiter.init(this.context);
     }
     TableRateLimiter.CreditFunction<?, ?> readCreditFn = deserializeObject(READ_CREDIT_FN);
-    TableRateLimiter readRateLimiter = new TableRateLimiter(tableSpec.getId(), rateLimiter, readCreditFn, RL_READ_TAG);
+    TableRateLimiter readRateLimiter = new TableRateLimiter(tableSpec.getId(), rateLimiter, readCreditFn, RemoteTableDescriptor.RL_READ_TAG);
 
     TableRateLimiter.CreditFunction<?, ?> writeCreditFn;
     TableRateLimiter writeRateLimiter = null;
@@ -112,7 +114,7 @@ public class RemoteTableProvider extends BaseTableProvider {
     boolean isRateLimited = readRateLimiter.isRateLimited();
     if (!readOnly) {
       writeCreditFn = deserializeObject(WRITE_CREDIT_FN);
-      writeRateLimiter = new TableRateLimiter(tableSpec.getId(), rateLimiter, writeCreditFn, RL_WRITE_TAG);
+      writeRateLimiter = new TableRateLimiter(tableSpec.getId(), rateLimiter, writeCreditFn, RemoteTableDescriptor.RL_WRITE_TAG);
       isRateLimited |= writeRateLimiter.isRateLimited();
       writeRetryPolicy = deserializeObject(WRITE_RETRY_POLICY);
       if (writeRetryPolicy != null) {
  * under the License.
  */
 
-package org.apache.samza.table.remote;
+package org.apache.samza.table.remote.descriptors;
 
-import org.apache.samza.table.TableProvider;
-import org.apache.samza.table.TableProviderFactory;
+import org.apache.samza.table.descriptors.TableProvider;
+import org.apache.samza.table.descriptors.TableProviderFactory;
 import org.apache.samza.table.TableSpec;
 
 import com.google.common.base.Preconditions;
index 1adddc0..8f7aa7e 100644 (file)
@@ -39,7 +39,7 @@ import static org.apache.samza.table.retry.FailsafeAdapter.failsafe;
 /**
  * Wrapper for a {@link TableReadFunction} instance to add common retry
  * support with a {@link TableRetryPolicy}. This wrapper is created by
- * {@link org.apache.samza.table.remote.RemoteTableProvider} when a retry
+ * {@link org.apache.samza.table.remote.descriptors.RemoteTableProvider} when a retry
  * policy is specified together with the {@link TableReadFunction}.
  *
  * Actual retry mechanism is provided by the failsafe library. Retry is
index 2f3f062..ee7959a 100644 (file)
@@ -39,7 +39,7 @@ import static org.apache.samza.table.retry.FailsafeAdapter.failsafe;
 /**
  * Wrapper for a {@link TableWriteFunction} instance to add common retry
  * support with a {@link TableRetryPolicy}. This wrapper is created by
- * {@link org.apache.samza.table.remote.RemoteTableProvider} when a retry
+ * {@link org.apache.samza.table.remote.descriptors.RemoteTableProvider} when a retry
  * policy is specified together with the {@link TableWriteFunction}.
  *
  * Actual retry mechanism is provided by the failsafe library. Retry is
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.table.utils;
+package org.apache.samza.table.utils.descriptors;
 
 import java.util.HashMap;
 import java.util.Map;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.JavaTableConfig;
 import org.apache.samza.context.Context;
-import org.apache.samza.table.TableProvider;
+import org.apache.samza.table.descriptors.TableProvider;
 import org.apache.samza.table.TableSpec;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
index 218ba5d..87131d7 100644 (file)
@@ -64,11 +64,12 @@ public class StreamOperatorTask implements StreamTask, InitableTask, WindowableT
    * Initializes this task during startup.
    * <p>
    * Implementation: Initializes the runtime {@link OperatorImplGraph} according to user-defined {@link OperatorSpecGraph}.
-   * Users set the input and output streams and the task-wide context manager using {@link org.apache.samza.application.StreamApplicationDescriptor} APIs,
-   * and the logical transforms using the {@link org.apache.samza.operators.MessageStream} APIs. After the
-   * {@link org.apache.samza.application.StreamApplicationDescriptorImpl} is initialized once by the application, it then creates
-   * an immutable {@link OperatorSpecGraph} accordingly, which is passed in to this class to create the {@link OperatorImplGraph}
-   * corresponding to the logical DAG.
+   * Users set the input and output streams and the task-wide context manager using
+   * {@link org.apache.samza.application.descriptors.StreamApplicationDescriptor} APIs, and the logical transforms
+   * using the {@link org.apache.samza.operators.MessageStream} APIs. After the
+   * {@link org.apache.samza.application.descriptors.StreamApplicationDescriptorImpl} is initialized once by the
+   * application, it then creates an immutable {@link OperatorSpecGraph} accordingly, which is passed in to this
+   * class to create the {@link OperatorImplGraph} corresponding to the logical DAG.
    *
    * @param context allows initializing and accessing contextual data of this StreamTask
    * @throws Exception in case of initialization errors
index c312fac..b2297e1 100644 (file)
@@ -21,10 +21,10 @@ package org.apache.samza.task;
 import com.google.common.base.Preconditions;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.samza.SamzaException;
-import org.apache.samza.application.ApplicationDescriptor;
-import org.apache.samza.application.ApplicationDescriptorImpl;
-import org.apache.samza.application.StreamApplicationDescriptorImpl;
-import org.apache.samza.application.TaskApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.ApplicationDescriptor;
+import org.apache.samza.application.descriptors.ApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.TaskApplicationDescriptorImpl;
 import org.apache.samza.config.ConfigException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
index 929d6a4..ee6aff3 100644 (file)
@@ -19,7 +19,8 @@
 
 package org.apache.samza.job.local
 
-import org.apache.samza.application.{ApplicationDescriptorUtil, ApplicationUtil}
+import org.apache.samza.application.ApplicationUtil
+import org.apache.samza.application.descriptors.ApplicationDescriptorUtil
 import org.apache.samza.config.JobConfig._
 import org.apache.samza.config.ShellCommandConfig._
 import org.apache.samza.config.{Config, TaskConfigJava}
index ccd88b8..8b96c8a 100644 (file)
@@ -18,6 +18,8 @@
  */
 package org.apache.samza.application;
 
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
+
 /**
  * Test class of {@link StreamApplication} for unit tests
  */
index 9b590c4..ab91cee 100644 (file)
@@ -25,6 +25,8 @@ import org.apache.samza.config.Config;
 import org.apache.samza.config.ConfigException;
 import org.apache.samza.config.MapConfig;
 import org.apache.samza.config.TaskConfig;
+import org.apache.samza.application.descriptors.TaskApplicationDescriptor;
+import org.apache.samza.application.descriptors.TaskApplicationDescriptorImpl;
 import org.apache.samza.task.MockStreamTask;
 import org.junit.Test;
 
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.application;
+package org.apache.samza.application.descriptors;
 
 import com.google.common.collect.ImmutableList;
 import java.util.ArrayList;
@@ -25,21 +25,22 @@ import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.samza.SamzaException;
+import org.apache.samza.application.StreamApplication;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.JobConfig;
 import org.apache.samza.context.ApplicationContainerContextFactory;
 import org.apache.samza.context.ApplicationTaskContextFactory;
-import org.apache.samza.operators.BaseTableDescriptor;
+import org.apache.samza.system.descriptors.GenericInputDescriptor;
+import org.apache.samza.system.descriptors.GenericOutputDescriptor;
+import org.apache.samza.system.descriptors.InputDescriptor;
+import org.apache.samza.system.descriptors.GenericSystemDescriptor;
+import org.apache.samza.table.descriptors.BaseTableDescriptor;
 import org.apache.samza.operators.data.TestMessageEnvelope;
-import org.apache.samza.operators.descriptors.GenericInputDescriptor;
-import org.apache.samza.operators.descriptors.GenericOutputDescriptor;
-import org.apache.samza.operators.descriptors.GenericSystemDescriptor;
-import org.apache.samza.operators.descriptors.base.stream.InputDescriptor;
-import org.apache.samza.operators.descriptors.base.system.ExpandingInputDescriptorProvider;
-import org.apache.samza.operators.descriptors.base.system.SystemDescriptor;
-import org.apache.samza.operators.descriptors.base.system.TransformingInputDescriptorProvider;
-import org.apache.samza.operators.functions.InputTransformer;
-import org.apache.samza.operators.functions.StreamExpander;
+import org.apache.samza.system.descriptors.ExpandingInputDescriptorProvider;
+import org.apache.samza.system.descriptors.SystemDescriptor;
+import org.apache.samza.system.descriptors.TransformingInputDescriptorProvider;
+import org.apache.samza.system.descriptors.InputTransformer;
+import org.apache.samza.system.descriptors.StreamExpander;
 import org.apache.samza.operators.spec.InputOperatorSpec;
 import org.apache.samza.operators.spec.OperatorSpec;
 import org.apache.samza.operators.spec.OperatorSpec.OpCode;
@@ -76,7 +77,7 @@ public class TestStreamApplicationDescriptorImpl {
     Config mockConfig = mock(Config.class);
     StreamApplicationDescriptorImpl appDesc = new StreamApplicationDescriptorImpl(mockApp, mockConfig);
     verify(mockApp).describe(appDesc);
-    assertEquals(mockConfig, appDesc.config);
+    assertEquals(mockConfig, appDesc.getConfig());
   }
 
   @Test
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.application;
+package org.apache.samza.application.descriptors;
 
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Optional;
 import java.util.Set;
+import org.apache.samza.application.TaskApplication;
 import org.apache.samza.config.Config;
 import org.apache.samza.context.ApplicationContainerContextFactory;
 import org.apache.samza.context.ApplicationTaskContextFactory;
-import org.apache.samza.operators.BaseTableDescriptor;
-import org.apache.samza.operators.TableDescriptor;
-import org.apache.samza.operators.descriptors.base.stream.InputDescriptor;
-import org.apache.samza.operators.descriptors.base.stream.OutputDescriptor;
-import org.apache.samza.operators.descriptors.base.system.SystemDescriptor;
+import org.apache.samza.system.descriptors.InputDescriptor;
+import org.apache.samza.system.descriptors.OutputDescriptor;
+import org.apache.samza.table.descriptors.BaseTableDescriptor;
+import org.apache.samza.table.descriptors.TableDescriptor;
+import org.apache.samza.system.descriptors.SystemDescriptor;
 import org.apache.samza.runtime.ProcessorLifecycleListenerFactory;
 import org.apache.samza.serializers.KVSerde;
 import org.apache.samza.task.TaskFactory;
@@ -90,7 +91,7 @@ public class TestTaskApplicationDescriptorImpl {
     TaskApplication mockApp = mock(TaskApplication.class);
     TaskApplicationDescriptorImpl appDesc = new TaskApplicationDescriptorImpl(mockApp, config);
     verify(mockApp).describe(appDesc);
-    assertEquals(config, appDesc.config);
+    assertEquals(config, appDesc.getConfig());
   }
 
   @Test
index 6308589..a5fde45 100644 (file)
@@ -21,20 +21,20 @@ package org.apache.samza.execution;
 import java.time.Duration;
 import java.util.HashMap;
 import java.util.Map;
-import org.apache.samza.application.ApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.ApplicationDescriptorImpl;
 import org.apache.samza.application.LegacyTaskApplication;
 import org.apache.samza.application.StreamApplication;
-import org.apache.samza.application.StreamApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptorImpl;
 import org.apache.samza.application.TaskApplication;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.JobConfig;
 import org.apache.samza.config.MapConfig;
+import org.apache.samza.system.descriptors.GenericInputDescriptor;
+import org.apache.samza.system.descriptors.GenericOutputDescriptor;
+import org.apache.samza.system.descriptors.GenericSystemDescriptor;
 import org.apache.samza.operators.KV;
 import org.apache.samza.operators.MessageStream;
 import org.apache.samza.operators.OutputStream;
-import org.apache.samza.operators.descriptors.GenericInputDescriptor;
-import org.apache.samza.operators.descriptors.GenericOutputDescriptor;
-import org.apache.samza.operators.descriptors.GenericSystemDescriptor;
 import org.apache.samza.operators.functions.JoinFunction;
 import org.apache.samza.serializers.JsonSerdeV2;
 import org.apache.samza.serializers.KVSerde;
index 4cfcfd2..6208206 100644 (file)
@@ -32,26 +32,26 @@ import java.util.Set;
 import java.util.stream.Collectors;
 import org.apache.samza.Partition;
 import org.apache.samza.SamzaException;
-import org.apache.samza.application.ApplicationDescriptor;
+import org.apache.samza.application.descriptors.ApplicationDescriptor;
 import org.apache.samza.application.LegacyTaskApplication;
 import org.apache.samza.application.SamzaApplication;
-import org.apache.samza.application.StreamApplicationDescriptorImpl;
-import org.apache.samza.application.TaskApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.TaskApplicationDescriptorImpl;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.JobConfig;
 import org.apache.samza.config.MapConfig;
 import org.apache.samza.config.TaskConfig;
-import org.apache.samza.operators.BaseTableDescriptor;
+import org.apache.samza.system.descriptors.GenericInputDescriptor;
+import org.apache.samza.system.descriptors.GenericOutputDescriptor;
+import org.apache.samza.system.descriptors.InputDescriptor;
+import org.apache.samza.system.descriptors.OutputDescriptor;
+import org.apache.samza.system.descriptors.GenericSystemDescriptor;
+import org.apache.samza.table.descriptors.BaseTableDescriptor;
 import org.apache.samza.operators.KV;
 import org.apache.samza.operators.MessageStream;
 import org.apache.samza.operators.OutputStream;
-import org.apache.samza.operators.TableDescriptor;
-import org.apache.samza.operators.descriptors.GenericInputDescriptor;
-import org.apache.samza.operators.descriptors.GenericOutputDescriptor;
-import org.apache.samza.operators.descriptors.GenericSystemDescriptor;
-import org.apache.samza.operators.descriptors.base.stream.InputDescriptor;
-import org.apache.samza.operators.descriptors.base.stream.OutputDescriptor;
-import org.apache.samza.operators.descriptors.base.system.SystemDescriptor;
+import org.apache.samza.table.descriptors.TableDescriptor;
+import org.apache.samza.system.descriptors.SystemDescriptor;
 import org.apache.samza.operators.functions.JoinFunction;
 import org.apache.samza.operators.functions.StreamTableJoinFunction;
 import org.apache.samza.operators.windows.Windows;
index 4de0485..979be94 100644 (file)
@@ -23,7 +23,7 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import org.apache.samza.application.StreamApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptorImpl;
 import org.apache.samza.system.StreamSpec;
 import org.junit.Before;
 import org.junit.Test;
index 34c16b3..b47014d 100644 (file)
@@ -26,17 +26,17 @@ import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
-import org.apache.samza.application.StreamApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptorImpl;
 import org.apache.samza.config.ApplicationConfig;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.JobConfig;
 import org.apache.samza.config.MapConfig;
+import org.apache.samza.system.descriptors.GenericInputDescriptor;
+import org.apache.samza.system.descriptors.GenericOutputDescriptor;
+import org.apache.samza.system.descriptors.GenericSystemDescriptor;
 import org.apache.samza.operators.KV;
 import org.apache.samza.operators.MessageStream;
 import org.apache.samza.operators.OutputStream;
-import org.apache.samza.operators.descriptors.GenericInputDescriptor;
-import org.apache.samza.operators.descriptors.GenericOutputDescriptor;
-import org.apache.samza.operators.descriptors.GenericSystemDescriptor;
 import org.apache.samza.operators.functions.JoinFunction;
 import org.apache.samza.operators.windows.Windows;
 import org.apache.samza.serializers.JsonSerdeV2;
index 4618e52..dda0ee1 100644 (file)
@@ -19,8 +19,8 @@
 package org.apache.samza.execution;
 
 import com.google.common.base.Joiner;
-import org.apache.samza.application.StreamApplicationDescriptorImpl;
-import org.apache.samza.application.TaskApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.TaskApplicationDescriptorImpl;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.ConfigRewriter;
 import org.apache.samza.config.JobConfig;
@@ -29,10 +29,10 @@ import org.apache.samza.config.SerializerConfig;
 import org.apache.samza.config.TaskConfig;
 import org.apache.samza.config.TaskConfigJava;
 import org.apache.samza.context.Context;
-import org.apache.samza.operators.BaseTableDescriptor;
+import org.apache.samza.system.descriptors.GenericInputDescriptor;
+import org.apache.samza.table.descriptors.BaseTableDescriptor;
 import org.apache.samza.operators.KV;
-import org.apache.samza.operators.TableDescriptor;
-import org.apache.samza.operators.descriptors.GenericInputDescriptor;
+import org.apache.samza.table.descriptors.TableDescriptor;
 import org.apache.samza.operators.impl.store.TimestampedValueSerde;
 import org.apache.samza.serializers.JsonSerdeV2;
 import org.apache.samza.serializers.KVSerde;
@@ -41,8 +41,8 @@ import org.apache.samza.serializers.SerializableSerde;
 import org.apache.samza.serializers.StringSerde;
 import org.apache.samza.system.StreamSpec;
 import org.apache.samza.table.Table;
-import org.apache.samza.table.TableProvider;
-import org.apache.samza.table.TableProviderFactory;
+import org.apache.samza.table.descriptors.TableProvider;
+import org.apache.samza.table.descriptors.TableProviderFactory;
 import org.apache.samza.table.TableSpec;
 import org.junit.Test;
 
index 9ed57fa..5a9b634 100644 (file)
@@ -22,9 +22,9 @@ import com.google.common.collect.ImmutableList;
 import java.util.Collections;
 import java.util.List;
 import java.util.stream.Collectors;
-import org.apache.samza.application.ApplicationDescriptor;
-import org.apache.samza.application.ApplicationDescriptorImpl;
-import org.apache.samza.application.StreamApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.ApplicationDescriptor;
+import org.apache.samza.application.descriptors.ApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptorImpl;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.JobConfig;
 import org.apache.samza.config.JobCoordinatorConfig;
index 85921f4..56be9a0 100644 (file)
@@ -20,9 +20,9 @@ package org.apache.samza.execution;
 
 import java.util.Collections;
 import java.util.List;
-import org.apache.samza.application.ApplicationDescriptor;
-import org.apache.samza.application.ApplicationDescriptorImpl;
-import org.apache.samza.application.StreamApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.ApplicationDescriptor;
+import org.apache.samza.application.descriptors.ApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptorImpl;
 import org.apache.samza.config.ApplicationConfig;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.JobConfig;
index 1315912..b3b0a5d 100644 (file)
@@ -21,15 +21,15 @@ package org.apache.samza.operators;
 import com.google.common.collect.ImmutableSet;
 import org.apache.samza.Partition;
 import org.apache.samza.SamzaException;
-import org.apache.samza.application.StreamApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptorImpl;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.MapConfig;
 import org.apache.samza.context.Context;
 import org.apache.samza.context.MockContext;
+import org.apache.samza.system.descriptors.GenericInputDescriptor;
+import org.apache.samza.system.descriptors.GenericSystemDescriptor;
 import org.apache.samza.job.model.TaskModel;
 import org.apache.samza.metrics.MetricsRegistryMap;
-import org.apache.samza.operators.descriptors.GenericInputDescriptor;
-import org.apache.samza.operators.descriptors.GenericSystemDescriptor;
 import org.apache.samza.operators.functions.JoinFunction;
 import org.apache.samza.operators.impl.store.TestInMemoryStore;
 import org.apache.samza.operators.impl.store.TimestampedValueSerde;
index 0f4784d..ce9435f 100644 (file)
@@ -24,7 +24,7 @@ import java.time.Duration;
 import java.util.Collection;
 import java.util.Collections;
 
-import org.apache.samza.application.StreamApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptorImpl;
 import org.apache.samza.operators.data.TestMessageEnvelope;
 import org.apache.samza.operators.data.TestOutputMessageEnvelope;
 import org.apache.samza.operators.functions.FilterFunction;
index abbbd3b..bbb70d2 100644 (file)
@@ -27,7 +27,7 @@ import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Set;
 import org.apache.samza.SamzaException;
-import org.apache.samza.application.StreamApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptorImpl;
 import org.apache.samza.operators.functions.ScheduledFunction;
 import org.apache.samza.operators.functions.WatermarkFunction;
 import org.apache.samza.operators.spec.InputOperatorSpec;
index d760805..6bbd674 100644 (file)
@@ -22,7 +22,7 @@ package org.apache.samza.operators.impl;
 import com.google.common.collect.HashMultimap;
 import com.google.common.collect.Multimap;
 import org.apache.samza.Partition;
-import org.apache.samza.application.StreamApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptorImpl;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.JobConfig;
 import org.apache.samza.config.MapConfig;
@@ -31,6 +31,9 @@ import org.apache.samza.container.TaskName;
 import org.apache.samza.context.Context;
 import org.apache.samza.context.MockContext;
 import org.apache.samza.context.TaskContextImpl;
+import org.apache.samza.system.descriptors.GenericInputDescriptor;
+import org.apache.samza.system.descriptors.GenericOutputDescriptor;
+import org.apache.samza.system.descriptors.GenericSystemDescriptor;
 import org.apache.samza.job.model.ContainerModel;
 import org.apache.samza.job.model.JobModel;
 import org.apache.samza.job.model.TaskModel;
@@ -38,9 +41,6 @@ import org.apache.samza.metrics.MetricsRegistryMap;
 import org.apache.samza.operators.KV;
 import org.apache.samza.operators.MessageStream;
 import org.apache.samza.operators.OutputStream;
-import org.apache.samza.operators.descriptors.GenericInputDescriptor;
-import org.apache.samza.operators.descriptors.GenericOutputDescriptor;
-import org.apache.samza.operators.descriptors.GenericSystemDescriptor;
 import org.apache.samza.operators.functions.ClosableFunction;
 import org.apache.samza.operators.functions.FilterFunction;
 import org.apache.samza.operators.functions.InitableFunction;
index 20d5e25..eae1ef4 100644 (file)
@@ -31,20 +31,20 @@ import java.util.List;
 import java.util.Map;
 import org.apache.samza.Partition;
 import org.apache.samza.application.StreamApplication;
-import org.apache.samza.application.StreamApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptorImpl;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.MapConfig;
 import org.apache.samza.container.TaskName;
 import org.apache.samza.context.Context;
 import org.apache.samza.context.MockContext;
 import org.apache.samza.context.TaskContextImpl;
+import org.apache.samza.system.descriptors.GenericInputDescriptor;
+import org.apache.samza.system.descriptors.GenericSystemDescriptor;
 import org.apache.samza.job.model.TaskModel;
 import org.apache.samza.metrics.MetricsRegistryMap;
 import org.apache.samza.operators.KV;
 import org.apache.samza.operators.MessageStream;
 import org.apache.samza.operators.OperatorSpecGraph;
-import org.apache.samza.operators.descriptors.GenericInputDescriptor;
-import org.apache.samza.operators.descriptors.GenericSystemDescriptor;
 import org.apache.samza.operators.functions.MapFunction;
 import org.apache.samza.operators.impl.store.TestInMemoryStore;
 import org.apache.samza.operators.impl.store.TimeSeriesKeySerde;
index fd4a7fb..64609b0 100644 (file)
@@ -20,14 +20,14 @@ package org.apache.samza.operators.spec;
 
 import java.util.Collection;
 import java.util.Map;
-import org.apache.samza.application.StreamApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptorImpl;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.JobConfig;
+import org.apache.samza.system.descriptors.GenericInputDescriptor;
+import org.apache.samza.system.descriptors.GenericSystemDescriptor;
 import org.apache.samza.operators.MessageStream;
 import org.apache.samza.operators.OperatorSpecGraph;
 import org.apache.samza.operators.Scheduler;
-import org.apache.samza.operators.descriptors.GenericInputDescriptor;
-import org.apache.samza.operators.descriptors.GenericSystemDescriptor;
 import org.apache.samza.operators.functions.MapFunction;
 import org.apache.samza.operators.functions.ScheduledFunction;
 import org.apache.samza.operators.functions.WatermarkFunction;
index fd0ddf8..1b93072 100644 (file)
@@ -23,11 +23,11 @@ import java.time.Duration;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
-import org.apache.samza.application.ApplicationDescriptor;
-import org.apache.samza.application.ApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.ApplicationDescriptor;
+import org.apache.samza.application.descriptors.ApplicationDescriptorImpl;
 import org.apache.samza.application.LegacyTaskApplication;
 import org.apache.samza.application.SamzaApplication;
-import org.apache.samza.application.ApplicationDescriptorUtil;
+import org.apache.samza.application.descriptors.ApplicationDescriptorUtil;
 import org.apache.samza.application.StreamApplication;
 import org.apache.samza.config.ApplicationConfig;
 import org.apache.samza.config.Config;
index 0952a87..cf23766 100644 (file)
@@ -24,6 +24,8 @@ import org.apache.samza.config.JavaTableConfig;
 import org.apache.samza.config.MapConfig;
 import org.apache.samza.config.SerializerConfig;
 import org.apache.samza.context.MockContext;
+import org.apache.samza.table.descriptors.TableProvider;
+import org.apache.samza.table.descriptors.TableProviderFactory;
 import org.apache.samza.serializers.IntegerSerde;
 import org.apache.samza.serializers.Serde;
 import org.apache.samza.serializers.SerializableSerde;
index 128b938..865c611 100644 (file)
@@ -28,15 +28,17 @@ import org.apache.samza.metrics.Counter;
 import org.apache.samza.metrics.Gauge;
 import org.apache.samza.metrics.MetricsRegistry;
 import org.apache.samza.metrics.Timer;
-import org.apache.samza.operators.BaseTableDescriptor;
-import org.apache.samza.operators.TableDescriptor;
+import org.apache.samza.table.descriptors.BaseTableDescriptor;
+import org.apache.samza.table.descriptors.TableDescriptor;
 import org.apache.samza.storage.kv.Entry;
 import org.apache.samza.table.ReadWriteTable;
 import org.apache.samza.table.ReadableTable;
 import org.apache.samza.table.TableSpec;
 import org.apache.samza.table.caching.guava.GuavaCacheTable;
-import org.apache.samza.table.caching.guava.GuavaCacheTableDescriptor;
-import org.apache.samza.table.caching.guava.GuavaCacheTableProvider;
+import org.apache.samza.table.caching.descriptors.CachingTableDescriptor;
+import org.apache.samza.table.caching.descriptors.CachingTableProvider;
+import org.apache.samza.table.caching.guava.descriptors.GuavaCacheTableDescriptor;
+import org.apache.samza.table.caching.guava.descriptors.GuavaCacheTableProvider;
 import org.apache.samza.table.remote.RemoteReadWriteTable;
 import org.apache.samza.table.remote.TableRateLimiter;
 import org.apache.samza.table.remote.TableReadFunction;
@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.apache.samza.table.remote;
+package org.apache.samza.table.remote.descriptors;
 
 import com.google.common.collect.ImmutableMap;
 import org.apache.samza.container.TaskName;
@@ -30,6 +30,11 @@ import org.apache.samza.metrics.MetricsRegistry;
 import org.apache.samza.metrics.Timer;
 import org.apache.samza.table.Table;
 import org.apache.samza.table.TableSpec;
+import org.apache.samza.table.remote.RemoteReadWriteTable;
+import org.apache.samza.table.remote.TableRateLimiter;
+import org.apache.samza.table.remote.TableReadFunction;
+import org.apache.samza.table.remote.TableWriteFunction;
+
 import org.apache.samza.table.retry.RetriableReadFunction;
 import org.apache.samza.table.retry.RetriableWriteFunction;
 import org.apache.samza.table.retry.TableRetryPolicy;
@@ -37,14 +42,12 @@ import org.apache.samza.util.EmbeddedTaggedRateLimiter;
 import org.apache.samza.util.RateLimiter;
 import org.junit.Assert;
 import org.junit.Test;
+import org.mockito.Matchers;
 
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ThreadPoolExecutor;
 
-import static org.apache.samza.table.remote.RemoteTableDescriptor.RL_READ_TAG;
-import static org.apache.samza.table.remote.RemoteTableDescriptor.RL_WRITE_TAG;
-import static org.mockito.Matchers.anyString;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
@@ -124,8 +127,8 @@ public class TestRemoteTableDescriptor {
     Context context = new MockContext();
 
     MetricsRegistry metricsRegistry = mock(MetricsRegistry.class);
-    doReturn(mock(Timer.class)).when(metricsRegistry).newTimer(anyString(), anyString());
-    doReturn(mock(Counter.class)).when(metricsRegistry).newCounter(anyString(), anyString());
+    doReturn(mock(Timer.class)).when(metricsRegistry).newTimer(Matchers.anyString(), Matchers.anyString());
+    doReturn(mock(Counter.class)).when(metricsRegistry).newCounter(Matchers.anyString(), Matchers.anyString());
     doReturn(metricsRegistry).when(context.getTaskContext()).getTaskMetricsRegistry();
 
     TaskName taskName = new TaskName("MyTask");
@@ -169,10 +172,10 @@ public class TestRemoteTableDescriptor {
       if (numRateLimitOps > 0) {
         Map<String, Integer> tagCredits = new HashMap<>();
         if (rlGets) {
-          tagCredits.put(RL_READ_TAG, 1000);
+          tagCredits.put(RemoteTableDescriptor.RL_READ_TAG, 1000);
         }
         if (rlPuts) {
-          tagCredits.put(RL_WRITE_TAG, 2000);
+          tagCredits.put(RemoteTableDescriptor.RL_WRITE_TAG, 2000);
         }
 
         // Spy the rate limiter to verify call count
@@ -188,15 +191,15 @@ public class TestRemoteTableDescriptor {
     Assert.assertTrue(table instanceof RemoteReadWriteTable);
     RemoteReadWriteTable rwTable = (RemoteReadWriteTable) table;
     if (numRateLimitOps > 0) {
-      Assert.assertTrue(!rlGets || rwTable.readRateLimiter != null);
-      Assert.assertTrue(!rlPuts || rwTable.writeRateLimiter != null);
+      Assert.assertTrue(!rlGets || rwTable.getReadRateLimiter() != null);
+      Assert.assertTrue(!rlPuts || rwTable.getWriteRateLimiter() != null);
     }
 
-    ThreadPoolExecutor callbackExecutor = (ThreadPoolExecutor) rwTable.callbackExecutor;
+    ThreadPoolExecutor callbackExecutor = (ThreadPoolExecutor) rwTable.getCallbackExecutor();
     Assert.assertEquals(10, callbackExecutor.getCorePoolSize());
 
-    Assert.assertNotNull(rwTable.readFn instanceof RetriableReadFunction);
-    Assert.assertNotNull(!(rwTable.writeFn instanceof RetriableWriteFunction));
+    Assert.assertNotNull(rwTable.getReadFn() instanceof RetriableReadFunction);
+    Assert.assertNotNull(!(rwTable.getWriteFn() instanceof RetriableWriteFunction));
   }
 
   @Test
index f96ab19..82fb41c 100644 (file)
@@ -21,9 +21,9 @@ package org.apache.samza.task;
 import java.lang.reflect.Field;
 import java.util.concurrent.ExecutorService;
 import org.apache.samza.SamzaException;
-import org.apache.samza.application.ApplicationDescriptorImpl;
-import org.apache.samza.application.StreamApplicationDescriptorImpl;
-import org.apache.samza.application.TaskApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.ApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.TaskApplicationDescriptorImpl;
 import org.apache.samza.config.ConfigException;
 import org.apache.samza.operators.OperatorSpecGraph;
 import org.junit.Test;
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.system.kafka;
+package org.apache.samza.system.kafka.descriptors;
 
 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.system.descriptors.InputDescriptor;
+import org.apache.samza.system.descriptors.SystemDescriptor;
+import org.apache.samza.system.descriptors.InputTransformer;
 import org.apache.samza.serializers.Serde;
 
 /**
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.system.kafka;
+package org.apache.samza.system.kafka.descriptors;
 
-import org.apache.samza.operators.descriptors.base.stream.OutputDescriptor;
-import org.apache.samza.operators.descriptors.base.system.SystemDescriptor;
+import org.apache.samza.system.descriptors.OutputDescriptor;
+import org.apache.samza.system.descriptors.SystemDescriptor;
 import org.apache.samza.serializers.Serde;
 
 /**
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.system.kafka;
+package org.apache.samza.system.kafka.descriptors;
 
 import java.util.Collections;
 import java.util.HashMap;
@@ -25,11 +25,12 @@ 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.system.descriptors.InputDescriptor;
+import org.apache.samza.system.descriptors.OutputDescriptorProvider;
+import org.apache.samza.system.descriptors.SimpleInputDescriptorProvider;
+import org.apache.samza.system.descriptors.SystemDescriptor;
 import org.apache.samza.serializers.Serde;
+import org.apache.samza.system.kafka.KafkaSystemFactory;
 
 
 /**
index 6a03198..3e75468 100644 (file)
@@ -52,7 +52,6 @@ import org.junit.Before;
 import org.junit.Test;
 import org.mockito.Mockito;
 
-import static org.apache.samza.system.kafka.KafkaSystemDescriptor.*;
 import static org.junit.Assert.*;
 import static org.mockito.Mockito.*;
 
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.system.kafka;
+package org.apache.samza.system.kafka.descriptors;
 
 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.apache.samza.system.kafka.descriptors.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
 import org.junit.Test;
 
 import static org.junit.Assert.assertEquals;
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.system.kafka;
+package org.apache.samza.system.kafka.descriptors;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 
 import java.util.Map;
 import org.apache.samza.system.SystemStreamMetadata;
+import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
 import org.junit.Test;
 
 import static org.junit.Assert.assertEquals;
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.storage.kv.inmemory;
+package org.apache.samza.storage.kv.inmemory.descriptors;
 
 import java.util.HashMap;
 import java.util.Map;
 
 import org.apache.samza.serializers.KVSerde;
-import org.apache.samza.storage.kv.BaseLocalStoreBackedTableDescriptor;
+import org.apache.samza.storage.kv.descriptors.BaseLocalStoreBackedTableDescriptor;
 import org.apache.samza.table.TableSpec;
 
 
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.storage.kv.inmemory;
+package org.apache.samza.storage.kv.inmemory.descriptors;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -24,7 +24,8 @@ import java.util.Map;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.JavaTableConfig;
 import org.apache.samza.config.StorageConfig;
-import org.apache.samza.storage.kv.BaseLocalStoreBackedTableProvider;
+import org.apache.samza.storage.kv.descriptors.BaseLocalStoreBackedTableProvider;
+import org.apache.samza.storage.kv.inmemory.InMemoryKeyValueStorageEngineFactory;
 import org.apache.samza.table.TableSpec;
 
 
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.storage.kv.inmemory;
+package org.apache.samza.storage.kv.inmemory.descriptors;
 
-import org.apache.samza.table.TableProvider;
-import org.apache.samza.table.TableProviderFactory;
+import org.apache.samza.table.descriptors.TableProvider;
+import org.apache.samza.table.descriptors.TableProviderFactory;
 import org.apache.samza.table.TableSpec;
 
 /**
@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.apache.samza.storage.kv.inmemory;
+package org.apache.samza.storage.kv.inmemory.descriptors;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -27,7 +27,8 @@ import org.apache.samza.config.MapConfig;
 import org.apache.samza.config.StorageConfig;
 import org.apache.samza.serializers.IntegerSerde;
 import org.apache.samza.serializers.KVSerde;
-import org.apache.samza.table.TableProvider;
+import org.apache.samza.storage.kv.inmemory.InMemoryKeyValueStorageEngineFactory;
+import org.apache.samza.table.descriptors.TableProvider;
 import org.apache.samza.table.TableSpec;
 import org.junit.Test;
 
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.storage.kv;
+package org.apache.samza.storage.kv.descriptors;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -25,6 +25,7 @@ import org.apache.samza.config.ClusterManagerConfig;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.JavaTableConfig;
 import org.apache.samza.config.StorageConfig;
+import org.apache.samza.storage.kv.RocksDbKeyValueStorageEngineFactory;
 import org.apache.samza.table.TableSpec;
 
 
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.storage.kv;
+package org.apache.samza.storage.kv.descriptors;
 
-import org.apache.samza.table.TableProvider;
-import org.apache.samza.table.TableProviderFactory;
+import org.apache.samza.table.descriptors.TableProvider;
+import org.apache.samza.table.descriptors.TableProviderFactory;
 import org.apache.samza.table.TableSpec;
 
 
@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.apache.samza.storage.kv;
+package org.apache.samza.storage.kv.descriptors;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -27,7 +27,8 @@ import org.apache.samza.config.MapConfig;
 import org.apache.samza.config.StorageConfig;
 import org.apache.samza.serializers.IntegerSerde;
 import org.apache.samza.serializers.KVSerde;
-import org.apache.samza.table.TableProvider;
+import org.apache.samza.storage.kv.RocksDbKeyValueStorageEngineFactory;
+import org.apache.samza.table.descriptors.TableProvider;
 import org.apache.samza.table.TableSpec;
 import org.junit.Test;
 
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.storage.kv;
+package org.apache.samza.storage.kv.descriptors;
 
 import com.google.common.base.Preconditions;
 
 import java.util.List;
 import java.util.Map;
 
-import org.apache.samza.operators.BaseTableDescriptor;
+import org.apache.samza.table.descriptors.BaseTableDescriptor;
 import org.apache.samza.serializers.KVSerde;
 import org.apache.samza.storage.SideInputsProcessor;
 
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.samza.storage.kv;
+package org.apache.samza.storage.kv.descriptors;
 
 import com.google.common.base.Preconditions;
 import java.util.HashMap;
@@ -32,10 +32,12 @@ import org.apache.samza.config.JobConfig;
 import org.apache.samza.config.MapConfig;
 import org.apache.samza.config.StorageConfig;
 import org.apache.samza.context.Context;
+import org.apache.samza.storage.kv.KeyValueStore;
+import org.apache.samza.storage.kv.LocalStoreBackedReadWriteTable;
 import org.apache.samza.table.ReadableTable;
 import org.apache.samza.table.Table;
 import org.apache.samza.table.TableSpec;
-import org.apache.samza.table.utils.BaseTableProvider;
+import org.apache.samza.table.utils.descriptors.BaseTableProvider;
 import org.apache.samza.table.utils.SerdeUtils;
 
 
@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.apache.samza.storage.kv;
+package org.apache.samza.storage.kv.descriptors;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -30,7 +30,8 @@ import org.apache.samza.config.MapConfig;
 import org.apache.samza.config.StorageConfig;
 import org.apache.samza.context.Context;
 import org.apache.samza.context.TaskContext;
-import org.apache.samza.table.TableProvider;
+import org.apache.samza.storage.kv.KeyValueStore;
+import org.apache.samza.table.descriptors.TableProvider;
 import org.apache.samza.table.TableSpec;
 import org.apache.samza.util.NoOpMetricsRegistry;
 import org.junit.Test;
index a1c1bdd..7faff17 100644 (file)
@@ -21,7 +21,7 @@ package org.apache.samza.sql.impl;
 
 import org.apache.samza.SamzaException;
 import org.apache.samza.config.Config;
-import org.apache.samza.operators.TableDescriptor;
+import org.apache.samza.table.descriptors.TableDescriptor;
 import org.apache.samza.serializers.JsonSerdeV2;
 import org.apache.samza.serializers.KVSerde;
 import org.apache.samza.sql.data.SamzaSqlCompositeKey;
@@ -29,7 +29,7 @@ import org.apache.samza.sql.data.SamzaSqlRelMessage;
 import org.apache.samza.sql.interfaces.SqlIOResolver;
 import org.apache.samza.sql.interfaces.SqlIOResolverFactory;
 import org.apache.samza.sql.interfaces.SqlIOConfig;
-import org.apache.samza.storage.kv.RocksDbTableDescriptor;
+import org.apache.samza.storage.kv.descriptors.RocksDbTableDescriptor;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
index 3a73e09..8636736 100644 (file)
@@ -29,7 +29,7 @@ import org.apache.commons.lang.Validate;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.MapConfig;
 import org.apache.samza.config.StreamConfig;
-import org.apache.samza.operators.TableDescriptor;
+import org.apache.samza.table.descriptors.TableDescriptor;
 import org.apache.samza.system.SystemStream;
 
 
index fd1a2a8..1caefe6 100644 (file)
@@ -25,7 +25,7 @@ import java.util.List;
 import java.util.Set;
 import org.apache.calcite.rel.RelRoot;
 import org.apache.samza.application.StreamApplication;
-import org.apache.samza.application.StreamApplicationDescriptor;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 import org.apache.samza.sql.dsl.SamzaSqlDslConverter;
 import org.apache.samza.sql.translator.QueryTranslator;
 import org.slf4j.Logger;
index 435a2cc..94b2296 100644 (file)
@@ -25,14 +25,14 @@ import java.util.Optional;
 import org.apache.calcite.rel.core.TableModify;
 import org.apache.commons.lang.Validate;
 import org.apache.samza.SamzaException;
-import org.apache.samza.application.StreamApplicationDescriptor;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 import org.apache.samza.context.Context;
+import org.apache.samza.system.descriptors.GenericOutputDescriptor;
 import org.apache.samza.operators.KV;
 import org.apache.samza.operators.MessageStream;
 import org.apache.samza.operators.MessageStreamImpl;
-import org.apache.samza.operators.TableDescriptor;
-import org.apache.samza.operators.descriptors.DelegatingSystemDescriptor;
-import org.apache.samza.operators.descriptors.GenericOutputDescriptor;
+import org.apache.samza.table.descriptors.TableDescriptor;
+import org.apache.samza.system.descriptors.DelegatingSystemDescriptor;
 import org.apache.samza.operators.functions.MapFunction;
 import org.apache.samza.serializers.KVSerde;
 import org.apache.samza.serializers.NoOpSerde;
index 817f145..c9365cc 100644 (file)
@@ -31,14 +31,11 @@ import org.apache.calcite.rel.logical.LogicalFilter;
 import org.apache.calcite.rel.logical.LogicalJoin;
 import org.apache.calcite.rel.logical.LogicalProject;
 import org.apache.samza.SamzaException;
-import org.apache.samza.application.StreamApplicationDescriptor;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 import org.apache.samza.context.Context;
 import org.apache.samza.operators.KV;
-import org.apache.samza.operators.functions.MapFunction;
 import org.apache.samza.operators.MessageStream;
-import org.apache.samza.operators.TableDescriptor;
-import org.apache.samza.operators.descriptors.DelegatingSystemDescriptor;
-import org.apache.samza.operators.descriptors.GenericOutputDescriptor;
+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.SamzaSqlExecutionContext;
@@ -50,7 +47,10 @@ import org.apache.samza.sql.planner.QueryPlanner;
 import org.apache.samza.sql.runner.SamzaSqlApplicationConfig;
 import org.apache.samza.sql.runner.SamzaSqlApplicationContext;
 import org.apache.samza.sql.testutil.SamzaSqlQueryParser;
+import org.apache.samza.system.descriptors.DelegatingSystemDescriptor;
+import org.apache.samza.system.descriptors.GenericOutputDescriptor;
 import org.apache.samza.table.Table;
+import org.apache.samza.table.descriptors.TableDescriptor;
 
 
 /**
index be94160..cc765bd 100644 (file)
@@ -23,12 +23,12 @@ import java.util.List;
 import java.util.Map;
 import org.apache.calcite.rel.core.TableScan;
 import org.apache.commons.lang.Validate;
-import org.apache.samza.application.StreamApplicationDescriptor;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 import org.apache.samza.context.Context;
+import org.apache.samza.system.descriptors.GenericInputDescriptor;
 import org.apache.samza.operators.KV;
 import org.apache.samza.operators.MessageStream;
-import org.apache.samza.operators.descriptors.DelegatingSystemDescriptor;
-import org.apache.samza.operators.descriptors.GenericInputDescriptor;
+import org.apache.samza.system.descriptors.DelegatingSystemDescriptor;
 import org.apache.samza.operators.functions.MapFunction;
 import org.apache.samza.serializers.KVSerde;
 import org.apache.samza.serializers.NoOpSerde;
index a7ab663..98cc92e 100644 (file)
@@ -32,9 +32,9 @@ import org.apache.calcite.rel.type.RelDataTypeFactory;
 import org.apache.calcite.rex.RexBuilder;
 import org.apache.calcite.rex.RexNode;
 import org.apache.calcite.schema.SchemaPlus;
-import org.apache.samza.application.StreamApplicationDescriptor;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 import org.apache.samza.operators.MessageStream;
-import org.apache.samza.operators.descriptors.DelegatingSystemDescriptor;
+import org.apache.samza.system.descriptors.DelegatingSystemDescriptor;
 import org.apache.samza.sql.data.RexToJavaCompiler;
 import org.apache.samza.sql.data.SamzaSqlExecutionContext;
 import org.apache.samza.sql.interfaces.SamzaRelConverter;
index 4c78b5a..14314c8 100644 (file)
@@ -26,8 +26,8 @@ import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import org.apache.commons.lang.NotImplementedException;
 import org.apache.samza.config.Config;
-import org.apache.samza.operators.BaseTableDescriptor;
-import org.apache.samza.operators.TableDescriptor;
+import org.apache.samza.table.descriptors.BaseTableDescriptor;
+import org.apache.samza.table.descriptors.TableDescriptor;
 import org.apache.samza.serializers.JsonSerdeV2;
 import org.apache.samza.serializers.KVSerde;
 import org.apache.samza.serializers.NoOpSerde;
@@ -36,13 +36,13 @@ import org.apache.samza.sql.data.SamzaSqlRelMessage;
 import org.apache.samza.sql.interfaces.SqlIOConfig;
 import org.apache.samza.sql.interfaces.SqlIOResolver;
 import org.apache.samza.sql.interfaces.SqlIOResolverFactory;
-import org.apache.samza.storage.kv.RocksDbTableDescriptor;
+import org.apache.samza.storage.kv.descriptors.RocksDbTableDescriptor;
 import org.apache.samza.table.ReadWriteTable;
 import org.apache.samza.table.Table;
-import org.apache.samza.table.TableProvider;
-import org.apache.samza.table.TableProviderFactory;
+import org.apache.samza.table.descriptors.TableProvider;
+import org.apache.samza.table.descriptors.TableProviderFactory;
 import org.apache.samza.table.TableSpec;
-import org.apache.samza.table.utils.BaseTableProvider;
+import org.apache.samza.table.utils.descriptors.BaseTableProvider;
 
 
 public class TestIOResolverFactory implements SqlIOResolverFactory {
index 07ebe33..b9b0c96 100644 (file)
@@ -24,7 +24,7 @@ import java.util.ArrayList;
 import org.apache.calcite.DataContext;
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.logical.LogicalFilter;
-import org.apache.samza.application.StreamApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptorImpl;
 import org.apache.samza.context.Context;
 import org.apache.samza.operators.MessageStream;
 import org.apache.samza.operators.MessageStreamImpl;
index f0a8a89..dcd7023 100644 (file)
@@ -33,10 +33,10 @@ import org.apache.calcite.rex.RexInputRef;
 import org.apache.calcite.rex.RexNode;
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.type.SqlTypeName;
-import org.apache.samza.application.StreamApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptorImpl;
 import org.apache.samza.operators.MessageStream;
 import org.apache.samza.operators.MessageStreamImpl;
-import org.apache.samza.operators.TableDescriptor;
+import org.apache.samza.table.descriptors.TableDescriptor;
 import org.apache.samza.operators.functions.StreamTableJoinFunction;
 import org.apache.samza.operators.spec.InputOperatorSpec;
 import org.apache.samza.operators.spec.OperatorSpec;
@@ -49,7 +49,7 @@ import org.apache.samza.sql.data.RexToJavaCompiler;
 import org.apache.samza.sql.data.SamzaSqlRelMessage;
 import org.apache.samza.sql.interfaces.SqlIOConfig;
 import org.apache.samza.sql.interfaces.SqlIOResolver;
-import org.apache.samza.storage.kv.RocksDbTableDescriptor;
+import org.apache.samza.storage.kv.descriptors.RocksDbTableDescriptor;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.internal.util.reflection.Whitebox;
index 2ed7a00..68db1e4 100644 (file)
@@ -31,7 +31,7 @@ import org.apache.calcite.rex.RexInputRef;
 import org.apache.calcite.rex.RexNode;
 import org.apache.calcite.sql.validate.SqlUserDefinedFunction;
 import org.apache.calcite.util.Pair;
-import org.apache.samza.application.StreamApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptorImpl;
 import org.apache.samza.context.Context;
 import org.apache.samza.operators.MessageStream;
 import org.apache.samza.operators.MessageStreamImpl;
index 7a194db..cd81e0d 100644 (file)
@@ -25,7 +25,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
 import org.apache.samza.SamzaException;
-import org.apache.samza.application.StreamApplicationDescriptorImpl;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptorImpl;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.MapConfig;
 import org.apache.samza.config.StreamConfig;
index a74993f..4943504 100644 (file)
@@ -22,14 +22,13 @@ package org.apache.samza.sql.translator;
 import java.util.HashMap;
 import java.util.Map;
 import org.apache.samza.operators.MessageStream;
-import org.apache.samza.operators.TableDescriptor;
+import org.apache.samza.table.descriptors.TableDescriptor;
 import org.apache.samza.operators.TableImpl;
 import org.apache.samza.serializers.JsonSerdeV2;
 import org.apache.samza.serializers.KVSerde;
 import org.apache.samza.serializers.StringSerde;
 import org.apache.samza.sql.data.SamzaSqlRelMessage;
-import org.apache.samza.storage.kv.RocksDbTableProvider;
-import org.apache.samza.table.Table;
+import org.apache.samza.storage.kv.descriptors.RocksDbTableProvider;
 import org.apache.samza.table.TableSpec;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
index 7d5e0d2..ba9c8b3 100644 (file)
@@ -21,7 +21,7 @@ package org.apache.samza.example;
 import java.time.Duration;
 import java.util.HashMap;
 import org.apache.samza.application.StreamApplication;
-import org.apache.samza.application.StreamApplicationDescriptor;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 import org.apache.samza.config.Config;
 import org.apache.samza.operators.KV;
 import org.apache.samza.operators.triggers.Triggers;
@@ -33,9 +33,9 @@ import org.apache.samza.runtime.ApplicationRunners;
 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.system.kafka.descriptors.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaOutputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
 import org.apache.samza.util.CommandLine;
 
 
index 4ef2402..7721d44 100644 (file)
@@ -20,7 +20,7 @@
 package org.apache.samza.example;
 
 import org.apache.samza.application.StreamApplication;
-import org.apache.samza.application.StreamApplicationDescriptor;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 import org.apache.samza.config.Config;
 import org.apache.samza.operators.KV;
 import org.apache.samza.operators.MessageStream;
@@ -29,9 +29,9 @@ import org.apache.samza.runtime.ApplicationRunners;
 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.system.kafka.descriptors.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaOutputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
 import org.apache.samza.util.CommandLine;
 
 
index e991c4e..4923b7d 100644 (file)
@@ -23,7 +23,7 @@ import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 import org.apache.samza.application.StreamApplication;
-import org.apache.samza.application.StreamApplicationDescriptor;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 import org.apache.samza.config.Config;
 import org.apache.samza.context.Context;
 import org.apache.samza.operators.KV;
@@ -36,9 +36,9 @@ 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.system.kafka.descriptors.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaOutputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
 import org.apache.samza.util.CommandLine;
 
 
index fe018f3..ac0db36 100644 (file)
@@ -22,7 +22,7 @@ package org.apache.samza.example;
 import com.google.common.collect.ImmutableList;
 
 import org.apache.samza.application.StreamApplication;
-import org.apache.samza.application.StreamApplicationDescriptor;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 import org.apache.samza.config.Config;
 import org.apache.samza.operators.MessageStream;
 import org.apache.samza.runtime.ApplicationRunner;
@@ -31,9 +31,9 @@ import org.apache.samza.operators.KV;
 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.system.kafka.descriptors.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaOutputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
 import org.apache.samza.util.CommandLine;
 
 public class MergeExample implements StreamApplication {
index 8d3812b..ea38984 100644 (file)
@@ -20,7 +20,7 @@ package org.apache.samza.example;
 
 import java.time.Duration;
 import org.apache.samza.application.StreamApplication;
-import org.apache.samza.application.StreamApplicationDescriptor;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 import org.apache.samza.config.Config;
 import org.apache.samza.operators.KV;
 import org.apache.samza.operators.functions.JoinFunction;
@@ -29,9 +29,9 @@ import org.apache.samza.runtime.ApplicationRunners;
 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.system.kafka.descriptors.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaOutputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
 import org.apache.samza.util.CommandLine;
 
 
index e2ebc93..1476c81 100644 (file)
@@ -19,7 +19,7 @@
 package org.apache.samza.example;
 
 import java.time.Duration;
-import org.apache.samza.application.StreamApplicationDescriptor;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 import org.apache.samza.application.StreamApplication;
 import org.apache.samza.config.Config;
 import org.apache.samza.operators.KV;
@@ -36,9 +36,9 @@ import org.apache.samza.runtime.ApplicationRunners;
 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.system.kafka.descriptors.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaOutputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
 import org.apache.samza.util.CommandLine;
 
 
index 8a0ca28..2cf3ac3 100644 (file)
@@ -20,7 +20,7 @@ package org.apache.samza.example;
 
 import java.time.Duration;
 import org.apache.samza.application.StreamApplication;
-import org.apache.samza.application.StreamApplicationDescriptor;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 import org.apache.samza.config.Config;
 import org.apache.samza.operators.KV;
 import org.apache.samza.operators.MessageStream;
@@ -32,9 +32,9 @@ import org.apache.samza.runtime.ApplicationRunners;
 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.system.kafka.descriptors.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaOutputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
 import org.apache.samza.util.CommandLine;
 
 
index 73dc10a..8f6c6f8 100644 (file)
  */
 package org.apache.samza.example;
 
-import org.apache.samza.application.TaskApplicationDescriptor;
+import org.apache.samza.application.descriptors.TaskApplicationDescriptor;
 import org.apache.samza.application.TaskApplication;
 import org.apache.samza.config.Config;
-import org.apache.samza.operators.TableDescriptor;
+import org.apache.samza.table.descriptors.TableDescriptor;
 import org.apache.samza.runtime.ApplicationRunner;
 import org.apache.samza.runtime.ApplicationRunners;
 import org.apache.samza.serializers.StringSerde;
-import org.apache.samza.storage.kv.RocksDbTableDescriptor;
+import org.apache.samza.storage.kv.descriptors.RocksDbTableDescriptor;
 import org.apache.samza.system.IncomingMessageEnvelope;
-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.system.kafka.descriptors.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaOutputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
 import org.apache.samza.task.MessageCollector;
 import org.apache.samza.task.StreamTask;
 import org.apache.samza.task.StreamTaskFactory;
index 2f4c19c..51089f7 100644 (file)
@@ -21,7 +21,7 @@ package org.apache.samza.example;
 
 import java.time.Duration;
 import org.apache.samza.application.StreamApplication;
-import org.apache.samza.application.StreamApplicationDescriptor;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 import org.apache.samza.config.Config;
 import org.apache.samza.operators.MessageStream;
 import org.apache.samza.operators.OutputStream;
@@ -34,9 +34,9 @@ import org.apache.samza.runtime.ApplicationRunner;
 import org.apache.samza.runtime.ApplicationRunners;
 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.system.kafka.descriptors.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaOutputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
 import org.apache.samza.util.CommandLine;
 
 
index 42379f3..b47bf0a 100644 (file)
@@ -24,7 +24,7 @@ import java.time.Duration;
 import java.util.stream.Collectors;
 import java.util.List;
 import java.util.Map;
-import org.apache.samza.test.framework.system.InMemoryOutputDescriptor;
+import org.apache.samza.test.framework.system.descriptors.InMemoryOutputDescriptor;
 import org.hamcrest.collection.IsIterableContainingInAnyOrder;
 import org.hamcrest.collection.IsIterableContainingInOrder;
 
@@ -32,8 +32,8 @@ import static org.junit.Assert.assertThat;
 
 
 /**
- * Assertion utils on the content of a stream described by
- * {@link org.apache.samza.operators.descriptors.base.stream.StreamDescriptor}.
+ * Assertion utils on the content of a stream described by a
+ * {@link org.apache.samza.system.descriptors.StreamDescriptor}
  */
 public class StreamAssert {
   /**
index 5cd47de..ba7128a 100644 (file)
@@ -58,9 +58,9 @@ import org.apache.samza.system.SystemStreamPartition;
 import org.apache.samza.system.inmemory.InMemorySystemFactory;
 import org.apache.samza.task.AsyncStreamTask;
 import org.apache.samza.task.StreamTask;
-import org.apache.samza.test.framework.system.InMemoryInputDescriptor;
-import org.apache.samza.test.framework.system.InMemoryOutputDescriptor;
-import org.apache.samza.test.framework.system.InMemorySystemDescriptor;
+import org.apache.samza.test.framework.system.descriptors.InMemoryInputDescriptor;
+import org.apache.samza.test.framework.system.descriptors.InMemoryOutputDescriptor;
+import org.apache.samza.test.framework.system.descriptors.InMemorySystemDescriptor;
 import org.apache.samza.util.FileUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -17,9 +17,9 @@
  * under the License.
  */
 
-package org.apache.samza.test.framework.system;
+package org.apache.samza.test.framework.system.descriptors;
 
-import org.apache.samza.operators.descriptors.base.stream.InputDescriptor;
+import org.apache.samza.system.descriptors.InputDescriptor;
 import org.apache.samza.serializers.NoOpSerde;
 
 /**
  * under the License.
  */
 
-package org.apache.samza.test.framework.system;
+package org.apache.samza.test.framework.system.descriptors;
 
-import org.apache.samza.operators.descriptors.base.stream.OutputDescriptor;
-import org.apache.samza.operators.descriptors.base.system.SystemDescriptor;
+import org.apache.samza.system.descriptors.OutputDescriptor;
+import org.apache.samza.system.descriptors.SystemDescriptor;
 import org.apache.samza.serializers.NoOpSerde;
 
 /**
  * under the License.
  */
 
-package org.apache.samza.test.framework.system;
+package org.apache.samza.test.framework.system.descriptors;
 
 import java.util.HashMap;
 import java.util.Map;
 import org.apache.samza.config.InMemorySystemConfig;
-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.system.descriptors.OutputDescriptorProvider;
+import org.apache.samza.system.descriptors.SimpleInputDescriptorProvider;
+import org.apache.samza.system.descriptors.SystemDescriptor;
 import org.apache.samza.serializers.Serde;
 import org.apache.samza.system.SystemStreamMetadata;
 import org.apache.samza.system.inmemory.InMemorySystemFactory;
index 1954cc3..2e51f6a 100644 (file)
 package org.apache.samza.test.integration;
 
 import org.apache.samza.application.StreamApplication;
-import org.apache.samza.application.StreamApplicationDescriptor;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 import org.apache.samza.operators.KV;
 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.apache.samza.system.kafka.descriptors.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaOutputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
index 4c8884d..672837b 100644 (file)
@@ -24,16 +24,16 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Random;
-import org.apache.samza.application.StreamApplicationDescriptor;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 import org.apache.samza.application.StreamApplication;
 import org.apache.samza.config.JobConfig;
 import org.apache.samza.config.JobCoordinatorConfig;
 import org.apache.samza.config.MapConfig;
 import org.apache.samza.config.TaskConfig;
 import org.apache.samza.container.grouper.task.SingleContainerGrouperFactory;
+import org.apache.samza.system.descriptors.GenericInputDescriptor;
 import org.apache.samza.operators.KV;
-import org.apache.samza.operators.descriptors.GenericInputDescriptor;
-import org.apache.samza.operators.descriptors.DelegatingSystemDescriptor;
+import org.apache.samza.system.descriptors.DelegatingSystemDescriptor;
 import org.apache.samza.runtime.ApplicationRunner;
 import org.apache.samza.runtime.ApplicationRunners;
 import org.apache.samza.serializers.KVSerde;
index e0097bd..ba62691 100644 (file)
@@ -29,7 +29,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import org.apache.samza.Partition;
-import org.apache.samza.application.StreamApplicationDescriptor;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 import org.apache.samza.application.StreamApplication;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.JobConfig;
@@ -40,10 +40,10 @@ import org.apache.samza.container.SamzaContainer;
 import org.apache.samza.container.TaskInstance;
 import org.apache.samza.container.TaskName;
 import org.apache.samza.container.grouper.task.SingleContainerGrouperFactory;
+import org.apache.samza.system.descriptors.GenericInputDescriptor;
 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.system.descriptors.DelegatingSystemDescriptor;
 import org.apache.samza.operators.impl.InputOperatorImpl;
 import org.apache.samza.operators.impl.OperatorImpl;
 import org.apache.samza.operators.impl.OperatorImplGraph;
index 7696b62..e326f91 100644 (file)
@@ -29,9 +29,9 @@ import java.util.stream.Collectors;
 import org.apache.samza.SamzaException;
 import org.apache.samza.operators.KV;
 import org.apache.samza.serializers.NoOpSerde;
-import org.apache.samza.test.framework.system.InMemoryInputDescriptor;
-import org.apache.samza.test.framework.system.InMemoryOutputDescriptor;
-import org.apache.samza.test.framework.system.InMemorySystemDescriptor;
+import org.apache.samza.test.framework.system.descriptors.InMemoryInputDescriptor;
+import org.apache.samza.test.framework.system.descriptors.InMemoryOutputDescriptor;
+import org.apache.samza.test.framework.system.descriptors.InMemorySystemDescriptor;
 import org.hamcrest.collection.IsIterableContainingInOrder;
 import org.junit.Assert;
 import org.junit.Test;
index 4caf266..ef17a22 100644 (file)
@@ -21,12 +21,12 @@ package org.apache.samza.test.framework;
 
 import java.util.Arrays;
 import org.apache.samza.application.StreamApplication;
-import org.apache.samza.application.StreamApplicationDescriptor;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 import org.apache.samza.config.Config;
 import org.apache.samza.operators.MessageStream;
 import org.apache.samza.serializers.JsonSerdeV2;
-import org.apache.samza.system.kafka.KafkaInputDescriptor;
-import org.apache.samza.system.kafka.KafkaSystemDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
 import org.apache.samza.test.operator.data.PageView;
 
 public class BroadcastAssertApp implements StreamApplication {
index 7deb4d7..649c032 100644 (file)
@@ -22,7 +22,7 @@ import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import org.apache.samza.application.TaskApplication;
-import org.apache.samza.application.TaskApplicationDescriptor;
+import org.apache.samza.application.descriptors.TaskApplicationDescriptor;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.JobConfig;
 import org.apache.samza.config.JobCoordinatorConfig;
@@ -31,8 +31,8 @@ import org.apache.samza.config.ZkConfig;
 import org.apache.samza.job.ApplicationStatus;
 import org.apache.samza.serializers.JsonSerdeV2;
 import org.apache.samza.system.IncomingMessageEnvelope;
-import org.apache.samza.system.kafka.KafkaInputDescriptor;
-import org.apache.samza.system.kafka.KafkaSystemDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
 import org.apache.samza.task.ClosableTask;
 import org.apache.samza.task.MessageCollector;
 import org.apache.samza.task.StreamTask;
index 6188381..a442140 100644 (file)
@@ -25,7 +25,7 @@ import java.util.List;
 import java.util.Random;
 import org.apache.samza.SamzaException;
 import org.apache.samza.application.StreamApplication;
-import org.apache.samza.application.StreamApplicationDescriptor;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 import org.apache.samza.operators.KV;
 import org.apache.samza.operators.MessageStream;
 import org.apache.samza.operators.OutputStream;
@@ -33,22 +33,23 @@ import org.apache.samza.serializers.IntegerSerde;
 import org.apache.samza.serializers.JsonSerdeV2;
 import org.apache.samza.serializers.KVSerde;
 import org.apache.samza.serializers.NoOpSerde;
-import org.apache.samza.storage.kv.RocksDbTableDescriptor;
+import org.apache.samza.storage.kv.descriptors.RocksDbTableDescriptor;
 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.KafkaOutputDescriptor;
-import org.apache.samza.system.kafka.KafkaSystemDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaOutputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
 import org.apache.samza.table.Table;
 import org.apache.samza.test.controlmessages.TestData;
-import org.apache.samza.test.framework.system.InMemoryInputDescriptor;
-import org.apache.samza.test.framework.system.InMemoryOutputDescriptor;
-import org.apache.samza.test.framework.system.InMemorySystemDescriptor;
+import org.apache.samza.test.framework.system.descriptors.InMemoryInputDescriptor;
+import org.apache.samza.test.framework.system.descriptors.InMemoryOutputDescriptor;
+import org.apache.samza.test.framework.system.descriptors.InMemorySystemDescriptor;
 import org.apache.samza.test.table.PageViewToProfileJoinFunction;
 import org.apache.samza.test.table.TestTableData;
 import org.junit.Assert;
 import org.junit.Test;
-import static org.apache.samza.test.controlmessages.TestData.*;
+
+import static org.apache.samza.test.controlmessages.TestData.PageView;
 
 public class StreamApplicationIntegrationTest {
 
index 003b200..aa9e107 100644 (file)
@@ -28,35 +28,36 @@ import java.util.Map;
 import java.util.stream.Collectors;
 import org.apache.samza.SamzaException;
 import org.apache.samza.application.TaskApplication;
-import org.apache.samza.application.TaskApplicationDescriptor;
 import org.apache.samza.context.Context;
+import org.apache.samza.application.descriptors.TaskApplicationDescriptor;
 import org.apache.samza.operators.KV;
 import org.apache.samza.serializers.IntegerSerde;
 import org.apache.samza.serializers.KVSerde;
 import org.apache.samza.serializers.NoOpSerde;
-import org.apache.samza.storage.kv.inmemory.InMemoryTableDescriptor;
+import org.apache.samza.storage.kv.inmemory.descriptors.InMemoryTableDescriptor;
 import org.apache.samza.system.IncomingMessageEnvelope;
 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.KafkaOutputDescriptor;
-import org.apache.samza.system.kafka.KafkaSystemDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaOutputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
 import org.apache.samza.table.ReadWriteTable;
 import org.apache.samza.task.InitableTask;
 import org.apache.samza.task.MessageCollector;
 import org.apache.samza.task.StreamTask;
 import org.apache.samza.task.StreamTaskFactory;
 import org.apache.samza.task.TaskCoordinator;
-import org.apache.samza.test.framework.system.InMemoryInputDescriptor;
-import org.apache.samza.test.framework.system.InMemoryOutputDescriptor;
-import org.apache.samza.test.framework.system.InMemorySystemDescriptor;
+import org.apache.samza.test.framework.system.descriptors.InMemoryInputDescriptor;
+import org.apache.samza.test.framework.system.descriptors.InMemoryOutputDescriptor;
+import org.apache.samza.test.framework.system.descriptors.InMemorySystemDescriptor;
 import org.apache.samza.test.table.TestTableData;
 import org.hamcrest.collection.IsIterableContainingInOrder;
 import org.junit.Assert;
 import org.junit.Test;
-import static org.apache.samza.test.table.TestTableData.Profile;
-import static org.apache.samza.test.table.TestTableData.PageView;
+
 import static org.apache.samza.test.table.TestTableData.EnrichedPageView;
+import static org.apache.samza.test.table.TestTableData.PageView;
+import static org.apache.samza.test.table.TestTableData.Profile;
 
 
 public class StreamTaskIntegrationTest {
index 1644a0f..20f18ee 100644 (file)
@@ -25,14 +25,14 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import org.apache.samza.application.StreamApplication;
-import org.apache.samza.application.StreamApplicationDescriptor;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 import org.apache.samza.operators.MessageStream;
 import org.apache.samza.operators.Scheduler;
 import org.apache.samza.operators.functions.FlatMapFunction;
 import org.apache.samza.operators.functions.ScheduledFunction;
 import org.apache.samza.serializers.JsonSerdeV2;
-import org.apache.samza.system.kafka.KafkaInputDescriptor;
-import org.apache.samza.system.kafka.KafkaSystemDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
 import org.apache.samza.test.operator.data.PageView;
 
 public class TestSchedulingApp implements StreamApplication {
index c63c11f..dda31ea 100644 (file)
@@ -23,7 +23,7 @@ import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 import org.apache.samza.application.StreamApplication;
-import org.apache.samza.application.StreamApplicationDescriptor;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 import org.apache.samza.config.Config;
 import org.apache.samza.operators.KV;
 import org.apache.samza.operators.MessageStream;
@@ -35,8 +35,8 @@ import org.apache.samza.serializers.KVSerde;
 import org.apache.samza.serializers.StringSerde;
 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.system.kafka.descriptors.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
 import org.apache.samza.task.TaskCoordinator;
 import org.apache.samza.test.operator.data.AdClick;
 import org.apache.samza.test.operator.data.PageView;
index 79a25e7..fdf0761 100644 (file)
@@ -21,16 +21,16 @@ package org.apache.samza.test.operator;
 
 import java.time.Duration;
 import org.apache.samza.application.StreamApplication;
-import org.apache.samza.application.StreamApplicationDescriptor;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 import org.apache.samza.operators.KV;
 import org.apache.samza.operators.windows.Windows;
 import org.apache.samza.serializers.IntegerSerde;
 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.system.kafka.descriptors.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaOutputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
 import org.apache.samza.test.operator.data.PageView;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
index f116f1d..6dd4303 100644 (file)
@@ -20,7 +20,7 @@
 package org.apache.samza.test.operator;
 
 import java.time.Duration;
-import org.apache.samza.application.StreamApplicationDescriptor;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 import org.apache.samza.application.StreamApplication;
 import org.apache.samza.config.Config;
 import org.apache.samza.operators.KV;
@@ -33,9 +33,9 @@ import org.apache.samza.serializers.IntegerSerde;
 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.system.kafka.descriptors.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaOutputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
 import org.apache.samza.test.operator.data.PageView;
 import org.apache.samza.util.CommandLine;
 
index 0184013..4b87169 100644 (file)
@@ -20,7 +20,7 @@
 package org.apache.samza.test.operator;
 
 import java.time.Duration;
-import org.apache.samza.application.StreamApplicationDescriptor;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 import org.apache.samza.application.StreamApplication;
 import org.apache.samza.config.Config;
 import org.apache.samza.operators.KV;
@@ -33,9 +33,9 @@ import org.apache.samza.serializers.IntegerSerde;
 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.system.kafka.descriptors.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaOutputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
 import org.apache.samza.test.operator.data.PageView;
 import org.apache.samza.util.CommandLine;
 
index 51f33b5..0991fa1 100644 (file)
@@ -23,7 +23,7 @@ import java.io.IOException;
 import java.io.ObjectInputStream;
 import java.io.Serializable;
 import java.util.concurrent.CountDownLatch;
-import org.apache.samza.application.StreamApplicationDescriptor;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 import org.apache.samza.application.StreamApplication;
 import org.apache.samza.config.ApplicationConfig;
 import org.apache.samza.config.Config;
@@ -33,9 +33,9 @@ import org.apache.samza.operators.OutputStream;
 import org.apache.samza.operators.functions.MapFunction;
 import org.apache.samza.serializers.NoOpSerde;
 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.system.kafka.descriptors.KafkaInputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaOutputDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
 
 
 /**
index d24cf57..a9741b4 100644 (file)
@@ -27,7 +27,7 @@ import java.util.List;
 import java.util.Map;
 import org.apache.samza.application.StreamApplication;
 import org.apache.samza.application.TaskApplication;
-import org.apache.samza.application.TaskApplicationDescriptor;
+import org.apache.samza.application.descriptors.TaskApplicationDescriptor;
 import org.apache.samza.config.JobConfig;
 import org.apache.samza.config.JobCoordinatorConfig;
 import org.apache.samza.config.MapConfig;
@@ -35,14 +35,14 @@ import org.apache.samza.config.TaskConfig;
 import org.apache.samza.container.grouper.task.SingleContainerGrouperFactory;
 import org.apache.samza.context.Context;
 import org.apache.samza.context.TaskContext;
+import org.apache.samza.system.descriptors.GenericInputDescriptor;
 import org.apache.samza.metrics.Counter;
 import org.apache.samza.metrics.Gauge;
 import org.apache.samza.metrics.MetricsRegistry;
 import org.apache.samza.metrics.Timer;
 import org.apache.samza.operators.KV;
 import org.apache.samza.operators.MessageStream;
-import org.apache.samza.operators.descriptors.DelegatingSystemDescriptor;
-import org.apache.samza.operators.descriptors.GenericInputDescriptor;
+import org.apache.samza.system.descriptors.DelegatingSystemDescriptor;
 import org.apache.samza.operators.functions.MapFunction;
 import org.apache.samza.runtime.LocalApplicationRunner;
 import org.apache.samza.serializers.IntegerSerde;
@@ -52,7 +52,7 @@ import org.apache.samza.standalone.PassthroughJobCoordinatorFactory;
 import org.apache.samza.storage.kv.Entry;
 import org.apache.samza.storage.kv.KeyValueStore;
 import org.apache.samza.storage.kv.LocalStoreBackedReadWriteTable;
-import org.apache.samza.storage.kv.inmemory.InMemoryTableDescriptor;
+import org.apache.samza.storage.kv.inmemory.descriptors.InMemoryTableDescriptor;
 import org.apache.samza.system.IncomingMessageEnvelope;
 import org.apache.samza.table.ReadWriteTable;
 import org.apache.samza.table.ReadableTable;
index 4410b87..5852de5 100644 (file)
@@ -28,23 +28,23 @@ import java.util.Map;
 import java.util.stream.Collectors;
 import org.apache.samza.SamzaException;
 import org.apache.samza.application.StreamApplication;
-import org.apache.samza.application.StreamApplicationDescriptor;
 import org.apache.samza.config.MapConfig;
 import org.apache.samza.config.StreamConfig;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 import org.apache.samza.operators.KV;
-import org.apache.samza.operators.TableDescriptor;
+import org.apache.samza.table.descriptors.TableDescriptor;
 import org.apache.samza.serializers.IntegerSerde;
 import org.apache.samza.serializers.KVSerde;
 import org.apache.samza.serializers.NoOpSerde;
 import org.apache.samza.storage.kv.Entry;
-import org.apache.samza.storage.kv.RocksDbTableDescriptor;
-import org.apache.samza.storage.kv.inmemory.InMemoryTableDescriptor;
-import org.apache.samza.system.kafka.KafkaSystemDescriptor;
+import org.apache.samza.storage.kv.descriptors.RocksDbTableDescriptor;
+import org.apache.samza.storage.kv.inmemory.descriptors.InMemoryTableDescriptor;
+import org.apache.samza.system.kafka.descriptors.KafkaSystemDescriptor;
 import org.apache.samza.table.Table;
 import org.apache.samza.test.framework.TestRunner;
-import org.apache.samza.test.framework.system.InMemoryInputDescriptor;
-import org.apache.samza.test.framework.system.InMemoryOutputDescriptor;
-import org.apache.samza.test.framework.system.InMemorySystemDescriptor;
+import org.apache.samza.test.framework.system.descriptors.InMemoryInputDescriptor;
+import org.apache.samza.test.framework.system.descriptors.InMemoryOutputDescriptor;
+import org.apache.samza.test.framework.system.descriptors.InMemorySystemDescriptor;
 import org.apache.samza.test.harness.AbstractIntegrationTestHarness;
 import org.junit.Test;
 
index a42f2e6..98c4193 100644 (file)
@@ -35,25 +35,25 @@ import java.util.function.Function;
 import java.util.stream.Collectors;
 import org.apache.samza.SamzaException;
 import org.apache.samza.application.StreamApplication;
-import org.apache.samza.application.StreamApplicationDescriptor;
+import org.apache.samza.application.descriptors.StreamApplicationDescriptor;
 import org.apache.samza.config.MapConfig;
 import org.apache.samza.context.Context;
 import org.apache.samza.context.TaskContext;
+import org.apache.samza.system.descriptors.GenericInputDescriptor;
 import org.apache.samza.metrics.Counter;
 import org.apache.samza.metrics.MetricsRegistry;
 import org.apache.samza.metrics.Timer;
 import org.apache.samza.operators.KV;
-import org.apache.samza.operators.TableDescriptor;
-import org.apache.samza.operators.descriptors.DelegatingSystemDescriptor;
-import org.apache.samza.operators.descriptors.GenericInputDescriptor;
+import org.apache.samza.table.descriptors.TableDescriptor;
+import org.apache.samza.system.descriptors.DelegatingSystemDescriptor;
 import org.apache.samza.runtime.LocalApplicationRunner;
 import org.apache.samza.serializers.NoOpSerde;
 import org.apache.samza.table.Table;
-import org.apache.samza.table.caching.CachingTableDescriptor;
-import org.apache.samza.table.caching.guava.GuavaCacheTableDescriptor;
+import org.apache.samza.table.caching.descriptors.CachingTableDescriptor;
+import org.apache.samza.table.caching.guava.descriptors.GuavaCacheTableDescriptor;
 import org.apache.samza.table.remote.RemoteReadWriteTable;
 import org.apache.samza.table.remote.RemoteReadableTable;
-import org.apache.samza.table.remote.RemoteTableDescriptor;
+import org.apache.samza.table.remote.descriptors.RemoteTableDescriptor;
 import org.apache.samza.table.remote.TableRateLimiter;
 import org.apache.samza.table.remote.TableReadFunction;
 import org.apache.samza.table.remote.TableWriteFunction;
@@ -212,7 +212,7 @@ public class TestRemoteTable extends AbstractIntegrationTestHarness {
           : appDesc.getTable(inputTableDesc);
 
       DelegatingSystemDescriptor ksd = new DelegatingSystemDescriptor("test");
-      GenericInputDescriptor<TestTableData.PageView> isd = ksd.getInputDescriptor("PageView", new NoOpSerde<>());
+      GenericInputDescriptor<PageView> isd = ksd.getInputDescriptor("PageView", new NoOpSerde<>());
       appDesc.getInputStream(isd)
           .map(pv -> new KV<>(pv.getMemberId(), pv))
           .join(inputTable, new PageViewToProfileJoinFunction())
index f868fdc..f4c05e4 100644 (file)
@@ -32,17 +32,17 @@ import org.apache.samza.config.JavaStorageConfig;
 import org.apache.samza.config.JavaTableConfig;
 import org.apache.samza.config.JobConfig;
 import org.apache.samza.config.MapConfig;
-import org.apache.samza.operators.TableDescriptor;
+import org.apache.samza.table.descriptors.TableDescriptor;
 import org.apache.samza.serializers.KVSerde;
 import org.apache.samza.serializers.LongSerde;
 import org.apache.samza.serializers.StringSerde;
 import org.apache.samza.storage.kv.RocksDbKeyValueStorageEngineFactory;
-import org.apache.samza.storage.kv.RocksDbTableDescriptor;
-import org.apache.samza.storage.kv.RocksDbTableProviderFactory;
+import org.apache.samza.storage.kv.descriptors.RocksDbTableDescriptor;
+import org.apache.samza.storage.kv.descriptors.RocksDbTableProviderFactory;
 import org.apache.samza.table.TableConfigGenerator;
 import org.apache.samza.table.TableDescriptorsProvider;
-import org.apache.samza.table.remote.RemoteTableDescriptor;
-import org.apache.samza.table.remote.RemoteTableProviderFactory;
+import org.apache.samza.table.remote.descriptors.RemoteTableDescriptor;
+import org.apache.samza.table.remote.descriptors.RemoteTableProviderFactory;
 import org.apache.samza.table.remote.TableReadFunction;
 import org.apache.samza.util.RateLimiter;
 import org.apache.samza.util.Util;
index f7805fe..c9f29eb 100644 (file)
@@ -36,9 +36,9 @@ import org.apache.samza.config.JobCoordinatorConfig;
 import org.apache.samza.config.MapConfig;
 import org.apache.samza.config.SystemConfig;
 import org.apache.samza.config.TaskConfig;
+import org.apache.samza.system.descriptors.GenericInputDescriptor;
+import org.apache.samza.system.descriptors.GenericSystemDescriptor;
 import org.apache.samza.operators.MessageStream;
-import org.apache.samza.operators.descriptors.GenericInputDescriptor;
-import org.apache.samza.operators.descriptors.GenericSystemDescriptor;
 import org.apache.samza.operators.functions.MapFunction;
 import org.apache.samza.runtime.ApplicationRunner;
 import org.apache.samza.runtime.ApplicationRunners;