IGNITE-3949: Applied new HadoopClassLoader architecture.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/11b00873 Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/11b00873 Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/11b00873 Branch: refs/heads/ignite-1.6.8-hadoop Commit: 11b008730580ea6420356b5d8c74755f55b780ae Parents: b49fe40 Author: vozerov-gridgain <voze...@gridgain.com> Authored: Thu Sep 22 10:00:07 2016 +0300 Committer: vozerov-gridgain <voze...@gridgain.com> Committed: Thu Sep 22 10:00:15 2016 +0300 ---------------------------------------------------------------------- .../apache/ignite/internal/IgniteKernal.java | 3 +- .../processors/hadoop/HadoopClassLoader.java | 70 +- .../processors/hadoop/HadoopDefaultJobInfo.java | 156 + .../processors/hadoop/HadoopHelper.java | 28 +- .../processors/hadoop/HadoopNoopHelper.java | 16 +- .../internal/processors/igfs/IgfsImpl.java | 5 +- .../processors/igfs/IgfsKernalContextAware.java | 32 + modules/hadoop-impl/pom.xml | 6 + .../hadoop/fs/v1/IgniteHadoopFileSystem.java | 26 +- .../hadoop/fs/v2/IgniteHadoopFileSystem.java | 26 +- .../IgniteHadoopClientProtocolProvider.java | 2 +- .../processors/hadoop/HadoopAttributes.java | 168 - .../processors/hadoop/HadoopComponent.java | 62 - .../processors/hadoop/HadoopContext.java | 201 - .../processors/hadoop/HadoopDefaultJobInfo.java | 156 - .../internal/processors/hadoop/HadoopImpl.java | 134 - .../hadoop/HadoopMapReduceCounterGroup.java | 123 - .../hadoop/HadoopMapReduceCounters.java | 228 - .../processors/hadoop/HadoopProcessor.java | 223 - .../internal/processors/hadoop/HadoopSetup.java | 542 - .../hadoop/HadoopTaskCancelledException.java | 35 - .../internal/processors/hadoop/HadoopUtils.java | 364 - .../hadoop/counter/HadoopCounterAdapter.java | 129 - .../hadoop/counter/HadoopCountersImpl.java | 200 - .../hadoop/counter/HadoopLongCounter.java | 93 - .../counter/HadoopPerformanceCounter.java | 288 - .../HadoopBasicFileSystemFactoryDelegate.java | 162 - .../HadoopCachingFileSystemFactoryDelegate.java | 75 - .../HadoopDefaultFileSystemFactoryDelegate.java | 61 - ...doopFileSystemCounterWriterDelegateImpl.java | 107 - ...doopIgfsSecondaryFileSystemDelegateImpl.java | 469 - ...HadoopKerberosFileSystemFactoryDelegate.java | 112 - .../hadoop/fs/HadoopFileSystemCacheUtils.java | 242 - .../hadoop/fs/HadoopFileSystemsUtils.java | 51 - .../hadoop/fs/HadoopLazyConcurrentMap.java | 212 - .../hadoop/fs/HadoopLocalFileSystemV1.java | 39 - .../hadoop/fs/HadoopLocalFileSystemV2.java | 88 - .../processors/hadoop/fs/HadoopParameters.java | 94 - .../hadoop/fs/HadoopRawLocalFileSystem.java | 314 - .../processors/hadoop/igfs/HadoopIgfs.java | 202 - .../igfs/HadoopIgfsCommunicationException.java | 57 - .../processors/hadoop/igfs/HadoopIgfsEx.java | 93 - .../hadoop/igfs/HadoopIgfsFuture.java | 97 - .../hadoop/igfs/HadoopIgfsInProc.java | 510 - .../hadoop/igfs/HadoopIgfsInputStream.java | 629 - .../processors/hadoop/igfs/HadoopIgfsIo.java | 76 - .../processors/hadoop/igfs/HadoopIgfsIpcIo.java | 624 - .../hadoop/igfs/HadoopIgfsIpcIoListener.java | 36 - .../hadoop/igfs/HadoopIgfsJclLogger.java | 116 - .../hadoop/igfs/HadoopIgfsOutProc.java | 524 - .../hadoop/igfs/HadoopIgfsOutputStream.java | 201 - .../hadoop/igfs/HadoopIgfsProperties.java | 86 - .../hadoop/igfs/HadoopIgfsProxyInputStream.java | 337 - .../igfs/HadoopIgfsProxyOutputStream.java | 165 - ...fsSecondaryFileSystemPositionedReadable.java | 105 - .../hadoop/igfs/HadoopIgfsStreamDelegate.java | 96 - .../igfs/HadoopIgfsStreamEventListener.java | 39 - .../processors/hadoop/igfs/HadoopIgfsUtils.java | 174 - .../hadoop/igfs/HadoopIgfsWrapper.java | 552 - .../impl/HadoopMapReduceCounterGroup.java | 123 + .../hadoop/impl/HadoopMapReduceCounters.java | 228 + .../processors/hadoop/impl/HadoopUtils.java | 327 + .../HadoopBasicFileSystemFactoryDelegate.java | 164 + .../HadoopCachingFileSystemFactoryDelegate.java | 75 + .../HadoopDefaultFileSystemFactoryDelegate.java | 62 + ...doopFileSystemCounterWriterDelegateImpl.java | 108 + ...doopIgfsSecondaryFileSystemDelegateImpl.java | 472 + ...HadoopKerberosFileSystemFactoryDelegate.java | 112 + .../impl/fs/HadoopFileSystemCacheUtils.java | 242 + .../hadoop/impl/fs/HadoopFileSystemsUtils.java | 51 + .../hadoop/impl/fs/HadoopLazyConcurrentMap.java | 209 + .../hadoop/impl/fs/HadoopLocalFileSystemV1.java | 39 + .../hadoop/impl/fs/HadoopLocalFileSystemV2.java | 88 + .../hadoop/impl/fs/HadoopParameters.java | 94 + .../impl/fs/HadoopRawLocalFileSystem.java | 314 + .../processors/hadoop/impl/igfs/HadoopIgfs.java | 202 + .../igfs/HadoopIgfsCommunicationException.java | 57 + .../hadoop/impl/igfs/HadoopIgfsEx.java | 93 + .../hadoop/impl/igfs/HadoopIgfsFuture.java | 97 + .../hadoop/impl/igfs/HadoopIgfsInProc.java | 510 + .../hadoop/impl/igfs/HadoopIgfsInputStream.java | 629 + .../hadoop/impl/igfs/HadoopIgfsIo.java | 76 + .../hadoop/impl/igfs/HadoopIgfsIpcIo.java | 624 + .../impl/igfs/HadoopIgfsIpcIoListener.java | 36 + .../hadoop/impl/igfs/HadoopIgfsJclLogger.java | 116 + .../hadoop/impl/igfs/HadoopIgfsOutProc.java | 524 + .../impl/igfs/HadoopIgfsOutputStream.java | 201 + .../hadoop/impl/igfs/HadoopIgfsProperties.java | 86 + .../impl/igfs/HadoopIgfsProxyInputStream.java | 337 + .../impl/igfs/HadoopIgfsProxyOutputStream.java | 165 + ...fsSecondaryFileSystemPositionedReadable.java | 105 + .../impl/igfs/HadoopIgfsStreamDelegate.java | 96 + .../igfs/HadoopIgfsStreamEventListener.java | 39 + .../hadoop/impl/igfs/HadoopIgfsUtils.java | 174 + .../hadoop/impl/igfs/HadoopIgfsWrapper.java | 553 + .../hadoop/impl/proto/HadoopClientProtocol.java | 353 + .../hadoop/impl/v1/HadoopV1CleanupTask.java | 64 + .../hadoop/impl/v1/HadoopV1Counter.java | 106 + .../hadoop/impl/v1/HadoopV1MapTask.java | 122 + .../hadoop/impl/v1/HadoopV1OutputCollector.java | 137 + .../hadoop/impl/v1/HadoopV1Partitioner.java | 44 + .../hadoop/impl/v1/HadoopV1ReduceTask.java | 101 + .../hadoop/impl/v1/HadoopV1Reporter.java | 81 + .../hadoop/impl/v1/HadoopV1SetupTask.java | 56 + .../hadoop/impl/v1/HadoopV1Splitter.java | 102 + .../processors/hadoop/impl/v1/HadoopV1Task.java | 97 + .../processors/hadoop/impl/v2/HadoopDaemon.java | 126 + .../hadoop/impl/v2/HadoopExternalSplit.java | 89 + .../impl/v2/HadoopSerializationWrapper.java | 138 + .../impl/v2/HadoopShutdownHookManager.java | 98 + .../hadoop/impl/v2/HadoopSplitWrapper.java | 119 + .../hadoop/impl/v2/HadoopV2CleanupTask.java | 72 + .../hadoop/impl/v2/HadoopV2Context.java | 243 + .../hadoop/impl/v2/HadoopV2Counter.java | 88 + .../processors/hadoop/impl/v2/HadoopV2Job.java | 450 + .../impl/v2/HadoopV2JobResourceManager.java | 324 + .../hadoop/impl/v2/HadoopV2MapTask.java | 99 + .../hadoop/impl/v2/HadoopV2Partitioner.java | 44 + .../hadoop/impl/v2/HadoopV2ReduceTask.java | 91 + .../hadoop/impl/v2/HadoopV2SetupTask.java | 65 + .../hadoop/impl/v2/HadoopV2Splitter.java | 111 + .../processors/hadoop/impl/v2/HadoopV2Task.java | 185 + .../hadoop/impl/v2/HadoopV2TaskContext.java | 560 + .../impl/v2/HadoopWritableSerialization.java | 75 + .../hadoop/jobtracker/HadoopJobMetadata.java | 316 - .../hadoop/jobtracker/HadoopJobTracker.java | 1700 --- .../hadoop/message/HadoopMessage.java | 27 - .../hadoop/proto/HadoopClientProtocol.java | 349 - .../proto/HadoopProtocolJobCountersTask.java | 46 - .../proto/HadoopProtocolJobStatusTask.java | 82 - .../hadoop/proto/HadoopProtocolKillJobTask.java | 46 - .../proto/HadoopProtocolNextTaskIdTask.java | 36 - .../proto/HadoopProtocolSubmitJobTask.java | 59 - .../hadoop/proto/HadoopProtocolTaskAdapter.java | 120 - .../proto/HadoopProtocolTaskArguments.java | 84 - .../hadoop/shuffle/HadoopShuffle.java | 263 - .../hadoop/shuffle/HadoopShuffleAck.java | 92 - .../hadoop/shuffle/HadoopShuffleJob.java | 612 - .../hadoop/shuffle/HadoopShuffleMessage.java | 242 - .../HadoopConcurrentHashMultimap.java | 616 - .../shuffle/collections/HadoopHashMultimap.java | 176 - .../collections/HadoopHashMultimapBase.java | 211 - .../shuffle/collections/HadoopMultimap.java | 113 - .../shuffle/collections/HadoopMultimapBase.java | 435 - .../shuffle/collections/HadoopSkipList.java | 733 - .../shuffle/streams/HadoopDataInStream.java | 171 - .../shuffle/streams/HadoopDataOutStream.java | 130 - .../shuffle/streams/HadoopOffheapBuffer.java | 122 - .../HadoopEmbeddedTaskExecutor.java | 153 - .../taskexecutor/HadoopExecutorService.java | 234 - .../hadoop/taskexecutor/HadoopRunnableTask.java | 293 - .../taskexecutor/HadoopTaskExecutorAdapter.java | 59 - .../hadoop/taskexecutor/HadoopTaskState.java | 38 - .../hadoop/taskexecutor/HadoopTaskStatus.java | 116 - .../external/HadoopExternalTaskExecutor.java | 976 -- .../external/HadoopExternalTaskMetadata.java | 67 - .../external/HadoopJobInfoUpdateRequest.java | 113 - .../external/HadoopPrepareForJobRequest.java | 130 - .../external/HadoopProcessDescriptor.java | 149 - .../external/HadoopProcessStartedAck.java | 47 - .../external/HadoopTaskExecutionRequest.java | 114 - .../external/HadoopTaskFinishedMessage.java | 94 - .../child/HadoopChildProcessRunner.java | 461 - .../child/HadoopExternalProcessStarter.java | 301 - .../HadoopAbstractCommunicationClient.java | 96 - .../HadoopCommunicationClient.java | 72 - .../HadoopExternalCommunication.java | 1460 -- .../HadoopHandshakeTimeoutException.java | 42 - .../communication/HadoopIpcToNioAdapter.java | 248 - .../communication/HadoopMarshallerFilter.java | 86 - .../communication/HadoopMessageListener.java | 39 - .../HadoopTcpNioCommunicationClient.java | 93 - .../hadoop/v1/HadoopV1CleanupTask.java | 64 - .../processors/hadoop/v1/HadoopV1Counter.java | 106 - .../processors/hadoop/v1/HadoopV1MapTask.java | 122 - .../hadoop/v1/HadoopV1OutputCollector.java | 137 - .../hadoop/v1/HadoopV1Partitioner.java | 44 - .../hadoop/v1/HadoopV1ReduceTask.java | 101 - .../processors/hadoop/v1/HadoopV1Reporter.java | 81 - .../processors/hadoop/v1/HadoopV1SetupTask.java | 56 - .../processors/hadoop/v1/HadoopV1Splitter.java | 102 - .../processors/hadoop/v1/HadoopV1Task.java | 97 - .../processors/hadoop/v2/HadoopDaemon.java | 126 - .../hadoop/v2/HadoopExternalSplit.java | 89 - .../hadoop/v2/HadoopSerializationWrapper.java | 138 - .../hadoop/v2/HadoopShutdownHookManager.java | 98 - .../hadoop/v2/HadoopSplitWrapper.java | 119 - .../hadoop/v2/HadoopV2CleanupTask.java | 72 - .../processors/hadoop/v2/HadoopV2Context.java | 243 - .../processors/hadoop/v2/HadoopV2Counter.java | 88 - .../processors/hadoop/v2/HadoopV2Job.java | 450 - .../hadoop/v2/HadoopV2JobResourceManager.java | 323 - .../processors/hadoop/v2/HadoopV2MapTask.java | 99 - .../hadoop/v2/HadoopV2Partitioner.java | 44 - .../hadoop/v2/HadoopV2ReduceTask.java | 91 - .../processors/hadoop/v2/HadoopV2SetupTask.java | 65 - .../processors/hadoop/v2/HadoopV2Splitter.java | 111 - .../processors/hadoop/v2/HadoopV2Task.java | 185 - .../hadoop/v2/HadoopV2TaskContext.java | 560 - .../hadoop/v2/HadoopWritableSerialization.java | 75 - .../HadoopClientProtocolEmbeddedSelfTest.java | 35 - .../hadoop/HadoopClientProtocolSelfTest.java | 654 - .../hadoop/cache/HadoopTxConfigCacheTest.java | 42 - ...KerberosHadoopFileSystemFactorySelfTest.java | 126 - .../util/BasicUserNameMapperSelfTest.java | 133 - .../util/ChainedUserNameMapperSelfTest.java | 107 - .../util/KerberosUserNameMapperSelfTest.java | 99 - .../ignite/igfs/Hadoop1DualAbstractTest.java | 158 - .../igfs/Hadoop1OverIgfsDualAsyncTest.java | 30 - .../igfs/Hadoop1OverIgfsDualSyncTest.java | 30 - .../igfs/HadoopFIleSystemFactorySelfTest.java | 340 - .../HadoopIgfs20FileSystemAbstractSelfTest.java | 2040 --- ...Igfs20FileSystemLoopbackPrimarySelfTest.java | 74 - ...oopIgfs20FileSystemShmemPrimarySelfTest.java | 74 - .../igfs/HadoopIgfsDualAbstractSelfTest.java | 321 - .../igfs/HadoopIgfsDualAsyncSelfTest.java | 32 - .../ignite/igfs/HadoopIgfsDualSyncSelfTest.java | 32 - ...adoopIgfsSecondaryFileSystemTestAdapter.java | 153 - ...oopSecondaryFileSystemConfigurationTest.java | 579 - .../apache/ignite/igfs/IgfsEventsTestSuite.java | 285 - .../igfs/IgfsNearOnlyMultiNodeSelfTest.java | 223 - .../IgniteHadoopFileSystemAbstractSelfTest.java | 2432 --- .../IgniteHadoopFileSystemClientSelfTest.java | 212 - ...IgniteHadoopFileSystemHandshakeSelfTest.java | 389 - .../IgniteHadoopFileSystemIpcCacheSelfTest.java | 214 - .../IgniteHadoopFileSystemLoggerSelfTest.java | 298 - ...niteHadoopFileSystemLoggerStateSelfTest.java | 329 - ...adoopFileSystemLoopbackAbstractSelfTest.java | 46 - ...SystemLoopbackEmbeddedDualAsyncSelfTest.java | 33 - ...eSystemLoopbackEmbeddedDualSyncSelfTest.java | 33 - ...leSystemLoopbackEmbeddedPrimarySelfTest.java | 33 - ...SystemLoopbackEmbeddedSecondarySelfTest.java | 34 - ...SystemLoopbackExternalDualAsyncSelfTest.java | 33 - ...eSystemLoopbackExternalDualSyncSelfTest.java | 33 - ...leSystemLoopbackExternalPrimarySelfTest.java | 33 - ...SystemLoopbackExternalSecondarySelfTest.java | 34 - ...condaryFileSystemInitializationSelfTest.java | 214 - ...teHadoopFileSystemShmemAbstractSelfTest.java | 91 - ...ileSystemShmemEmbeddedDualAsyncSelfTest.java | 33 - ...FileSystemShmemEmbeddedDualSyncSelfTest.java | 33 - ...pFileSystemShmemEmbeddedPrimarySelfTest.java | 33 - ...ileSystemShmemEmbeddedSecondarySelfTest.java | 33 - ...ileSystemShmemExternalDualAsyncSelfTest.java | 33 - ...FileSystemShmemExternalDualSyncSelfTest.java | 33 - ...pFileSystemShmemExternalPrimarySelfTest.java | 33 - ...ileSystemShmemExternalSecondarySelfTest.java | 33 - .../hadoop/HadoopAbstractMapReduceTest.java | 429 - .../hadoop/HadoopAbstractSelfTest.java | 239 - .../hadoop/HadoopAbstractWordCountTest.java | 175 - .../hadoop/HadoopClassLoaderTest.java | 110 - .../hadoop/HadoopCommandLineTest.java | 474 - .../HadoopDefaultMapReducePlannerSelfTest.java | 615 - .../processors/hadoop/HadoopErrorSimulator.java | 326 - .../hadoop/HadoopFileSystemsTest.java | 155 - .../processors/hadoop/HadoopGroupingTest.java | 307 - .../hadoop/HadoopJobTrackerSelfTest.java | 345 - .../hadoop/HadoopMapReduceEmbeddedSelfTest.java | 253 - .../HadoopMapReduceErrorResilienceTest.java | 154 - .../processors/hadoop/HadoopMapReduceTest.java | 66 - .../hadoop/HadoopNoHadoopMapReduceTest.java | 47 - .../processors/hadoop/HadoopPlannerMockJob.java | 168 - .../hadoop/HadoopPopularWordsTest.java | 298 - .../HadoopSerializationWrapperSelfTest.java | 79 - .../processors/hadoop/HadoopSharedMap.java | 66 - .../hadoop/HadoopSnappyFullMapReduceTest.java | 36 - .../processors/hadoop/HadoopSnappyTest.java | 102 - .../hadoop/HadoopSortingExternalTest.java | 46 - .../processors/hadoop/HadoopSortingTest.java | 303 - .../hadoop/HadoopSplitWrapperSelfTest.java | 72 - .../processors/hadoop/HadoopStartup.java | 54 - .../hadoop/HadoopTaskExecutionSelfTest.java | 567 - .../hadoop/HadoopTasksAllVersionsTest.java | 260 - .../processors/hadoop/HadoopTasksV1Test.java | 58 - .../processors/hadoop/HadoopTasksV2Test.java | 77 - .../hadoop/HadoopTestRoundRobinMrPlanner.java | 71 - .../hadoop/HadoopTestTaskContext.java | 228 - .../processors/hadoop/HadoopTestUtils.java | 178 - .../hadoop/HadoopUserLibsSelfTest.java | 260 - .../processors/hadoop/HadoopV2JobSelfTest.java | 100 - .../hadoop/HadoopValidationSelfTest.java | 53 - .../HadoopWeightedMapReducePlannerTest.java | 599 - .../HadoopWeightedPlannerMapReduceTest.java | 38 - .../hadoop/books/alice-in-wonderland.txt | 3735 ----- .../processors/hadoop/books/art-of-war.txt | 6982 --------- .../hadoop/books/huckleberry-finn.txt | 11733 --------------- .../processors/hadoop/books/sherlock-holmes.txt | 13052 ----------------- .../processors/hadoop/books/tom-sawyer.txt | 8858 ----------- .../hadoop/deps/CircularWIthHadoop.java | 32 - .../hadoop/deps/CircularWithoutHadoop.java | 27 - .../processors/hadoop/deps/WithCast.java | 41 - .../hadoop/deps/WithClassAnnotation.java | 28 - .../hadoop/deps/WithConstructorInvocation.java | 31 - .../processors/hadoop/deps/WithExtends.java | 27 - .../processors/hadoop/deps/WithField.java | 29 - .../processors/hadoop/deps/WithImplements.java | 36 - .../hadoop/deps/WithIndirectField.java | 27 - .../processors/hadoop/deps/WithInitializer.java | 33 - .../processors/hadoop/deps/WithInnerClass.java | 31 - .../hadoop/deps/WithLocalVariable.java | 38 - .../hadoop/deps/WithMethodAnnotation.java | 32 - .../hadoop/deps/WithMethodArgument.java | 31 - .../hadoop/deps/WithMethodCheckedException.java | 31 - .../hadoop/deps/WithMethodInvocation.java | 31 - .../hadoop/deps/WithMethodReturnType.java | 31 - .../hadoop/deps/WithMethodRuntimeException.java | 31 - .../processors/hadoop/deps/WithOuterClass.java | 38 - .../hadoop/deps/WithParameterAnnotation.java | 31 - .../processors/hadoop/deps/WithStaticField.java | 29 - .../hadoop/deps/WithStaticInitializer.java | 34 - .../processors/hadoop/deps/Without.java | 25 - .../hadoop/examples/HadoopWordCount1.java | 94 - .../hadoop/examples/HadoopWordCount1Map.java | 79 - .../hadoop/examples/HadoopWordCount1Reduce.java | 61 - .../hadoop/examples/HadoopWordCount2.java | 111 - .../examples/HadoopWordCount2Combiner.java | 45 - .../hadoop/examples/HadoopWordCount2Mapper.java | 88 - .../examples/HadoopWordCount2Reducer.java | 113 - .../impl/HadoopAbstractMapReduceTest.java | 430 + .../hadoop/impl/HadoopAbstractSelfTest.java | 239 + .../impl/HadoopAbstractWordCountTest.java | 175 + .../hadoop/impl/HadoopCommandLineTest.java | 476 + .../HadoopDefaultMapReducePlannerSelfTest.java | 619 + .../hadoop/impl/HadoopErrorSimulator.java | 326 + .../hadoop/impl/HadoopFileSystemsTest.java | 155 + .../hadoop/impl/HadoopGroupingTest.java | 308 + .../hadoop/impl/HadoopJobTrackerSelfTest.java | 348 + .../impl/HadoopMapReduceEmbeddedSelfTest.java | 254 + .../HadoopMapReduceErrorResilienceTest.java | 154 + .../hadoop/impl/HadoopMapReduceTest.java | 66 + .../impl/HadoopNoHadoopMapReduceTest.java | 47 + .../hadoop/impl/HadoopPlannerMockJob.java | 175 + .../hadoop/impl/HadoopPopularWordsTest.java | 298 + .../HadoopSerializationWrapperSelfTest.java | 80 + .../processors/hadoop/impl/HadoopSharedMap.java | 66 + .../impl/HadoopSnappyFullMapReduceTest.java | 36 + .../hadoop/impl/HadoopSnappyTest.java | 104 + .../hadoop/impl/HadoopSortingExternalTest.java | 46 + .../hadoop/impl/HadoopSortingTest.java | 304 + .../hadoop/impl/HadoopSplitWrapperSelfTest.java | 72 + .../processors/hadoop/impl/HadoopStartup.java | 54 + .../impl/HadoopTaskExecutionSelfTest.java | 570 + .../hadoop/impl/HadoopTasksAllVersionsTest.java | 264 + .../hadoop/impl/HadoopTasksV1Test.java | 62 + .../hadoop/impl/HadoopTasksV2Test.java | 81 + .../impl/HadoopTestRoundRobinMrPlanner.java | 75 + .../hadoop/impl/HadoopTestTaskContext.java | 233 + .../processors/hadoop/impl/HadoopTestUtils.java | 178 + .../hadoop/impl/HadoopTxConfigCacheTest.java | 42 + .../hadoop/impl/HadoopUserLibsSelfTest.java | 261 + .../hadoop/impl/HadoopV2JobSelfTest.java | 108 + .../hadoop/impl/HadoopValidationSelfTest.java | 53 + .../HadoopWeightedMapReducePlannerTest.java | 602 + .../HadoopWeightedPlannerMapReduceTest.java | 38 + .../hadoop/impl/books/alice-in-wonderland.txt | 3735 +++++ .../processors/hadoop/impl/books/art-of-war.txt | 6982 +++++++++ .../hadoop/impl/books/huckleberry-finn.txt | 11733 +++++++++++++++ .../hadoop/impl/books/sherlock-holmes.txt | 13052 +++++++++++++++++ .../processors/hadoop/impl/books/tom-sawyer.txt | 8858 +++++++++++ .../HadoopClientProtocolEmbeddedSelfTest.java | 35 + .../client/HadoopClientProtocolSelfTest.java | 654 + .../hadoop/impl/examples/HadoopWordCount1.java | 94 + .../impl/examples/HadoopWordCount1Map.java | 79 + .../impl/examples/HadoopWordCount1Reduce.java | 61 + .../hadoop/impl/examples/HadoopWordCount2.java | 111 + .../impl/examples/HadoopWordCount2Combiner.java | 45 + .../impl/examples/HadoopWordCount2Mapper.java | 88 + .../impl/examples/HadoopWordCount2Reducer.java | 113 + ...KerberosHadoopFileSystemFactorySelfTest.java | 126 + .../impl/igfs/Hadoop1DualAbstractTest.java | 157 + .../impl/igfs/Hadoop1OverIgfsDualAsyncTest.java | 32 + .../impl/igfs/Hadoop1OverIgfsDualSyncTest.java | 32 + .../igfs/HadoopFIleSystemFactorySelfTest.java | 345 + .../HadoopIgfs20FileSystemAbstractSelfTest.java | 2047 +++ ...Igfs20FileSystemLoopbackPrimarySelfTest.java | 77 + ...oopIgfs20FileSystemShmemPrimarySelfTest.java | 77 + .../igfs/HadoopIgfsDualAbstractSelfTest.java | 328 + .../impl/igfs/HadoopIgfsDualAsyncSelfTest.java | 32 + .../impl/igfs/HadoopIgfsDualSyncSelfTest.java | 32 + ...adoopIgfsSecondaryFileSystemTestAdapter.java | 153 + ...oopSecondaryFileSystemConfigurationTest.java | 583 + .../hadoop/impl/igfs/IgfsEventsTestSuite.java | 289 + .../igfs/IgfsNearOnlyMultiNodeSelfTest.java | 226 + .../IgniteHadoopFileSystemAbstractSelfTest.java | 2435 +++ .../IgniteHadoopFileSystemClientSelfTest.java | 216 + ...IgniteHadoopFileSystemHandshakeSelfTest.java | 393 + .../IgniteHadoopFileSystemIpcCacheSelfTest.java | 215 + .../IgniteHadoopFileSystemLoggerSelfTest.java | 299 + ...niteHadoopFileSystemLoggerStateSelfTest.java | 332 + ...adoopFileSystemLoopbackAbstractSelfTest.java | 50 + ...SystemLoopbackEmbeddedDualAsyncSelfTest.java | 33 + ...eSystemLoopbackEmbeddedDualSyncSelfTest.java | 33 + ...leSystemLoopbackEmbeddedPrimarySelfTest.java | 33 + ...SystemLoopbackEmbeddedSecondarySelfTest.java | 34 + ...SystemLoopbackExternalDualAsyncSelfTest.java | 33 + ...eSystemLoopbackExternalDualSyncSelfTest.java | 33 + ...leSystemLoopbackExternalPrimarySelfTest.java | 33 + ...SystemLoopbackExternalSecondarySelfTest.java | 34 + ...condaryFileSystemInitializationSelfTest.java | 217 + ...teHadoopFileSystemShmemAbstractSelfTest.java | 94 + ...ileSystemShmemEmbeddedDualAsyncSelfTest.java | 33 + ...FileSystemShmemEmbeddedDualSyncSelfTest.java | 33 + ...pFileSystemShmemEmbeddedPrimarySelfTest.java | 33 + ...ileSystemShmemEmbeddedSecondarySelfTest.java | 33 + ...ileSystemShmemExternalDualAsyncSelfTest.java | 33 + ...FileSystemShmemExternalDualSyncSelfTest.java | 33 + ...pFileSystemShmemExternalPrimarySelfTest.java | 33 + ...ileSystemShmemExternalSecondarySelfTest.java | 33 + .../taskexecutor/HadoopExecutorServiceTest.java | 119 + .../impl/util/BasicUserNameMapperSelfTest.java | 134 + .../util/ChainedUserNameMapperSelfTest.java | 111 + .../util/KerberosUserNameMapperSelfTest.java | 100 + .../collections/HadoopAbstractMapTest.java | 2 +- .../collections/HadoopSkipListSelfTest.java | 4 +- .../streams/HadoopDataStreamSelfTest.java | 1 + .../taskexecutor/HadoopExecutorServiceTest.java | 118 - .../HadoopExternalTaskExecutionSelfTest.java | 4 +- .../testsuites/IgniteHadoopTestSuite.java | 103 +- .../testsuites/IgniteHadoopTestSuite2.java | 116 + .../IgniteIgfsLinuxAndMacOSTestSuite.java | 22 +- .../fs/IgniteHadoopFileSystemCounterWriter.java | 7 +- .../fs/IgniteHadoopIgfsSecondaryFileSystem.java | 31 +- .../processors/hadoop/HadoopAttributes.java | 168 + .../processors/hadoop/HadoopCommonUtils.java | 44 + .../processors/hadoop/HadoopComponent.java | 62 + .../processors/hadoop/HadoopContext.java | 200 + .../processors/hadoop/HadoopHelperImpl.java | 596 +- .../internal/processors/hadoop/HadoopImpl.java | 134 + .../processors/hadoop/HadoopProcessor.java | 230 + .../internal/processors/hadoop/HadoopSetup.java | 542 + .../hadoop/HadoopTaskCancelledException.java | 35 + .../hadoop/counter/HadoopCounterAdapter.java | 130 + .../hadoop/counter/HadoopCountersImpl.java | 200 + .../hadoop/counter/HadoopLongCounter.java | 93 + .../counter/HadoopPerformanceCounter.java | 286 + .../hadoop/delegate/HadoopDelegateUtils.java | 31 +- .../hadoop/jobtracker/HadoopJobMetadata.java | 316 + .../hadoop/jobtracker/HadoopJobTracker.java | 1699 +++ .../hadoop/message/HadoopMessage.java | 27 + .../proto/HadoopProtocolJobCountersTask.java | 46 + .../proto/HadoopProtocolJobStatusTask.java | 82 + .../hadoop/proto/HadoopProtocolKillJobTask.java | 46 + .../proto/HadoopProtocolNextTaskIdTask.java | 36 + .../proto/HadoopProtocolSubmitJobTask.java | 59 + .../hadoop/proto/HadoopProtocolTaskAdapter.java | 120 + .../proto/HadoopProtocolTaskArguments.java | 84 + .../hadoop/shuffle/HadoopShuffle.java | 263 + .../hadoop/shuffle/HadoopShuffleAck.java | 92 + .../hadoop/shuffle/HadoopShuffleJob.java | 612 + .../hadoop/shuffle/HadoopShuffleMessage.java | 242 + .../HadoopConcurrentHashMultimap.java | 616 + .../shuffle/collections/HadoopHashMultimap.java | 176 + .../collections/HadoopHashMultimapBase.java | 211 + .../shuffle/collections/HadoopMultimap.java | 113 + .../shuffle/collections/HadoopMultimapBase.java | 435 + .../shuffle/collections/HadoopSkipList.java | 733 + .../shuffle/streams/HadoopDataInStream.java | 171 + .../shuffle/streams/HadoopDataOutStream.java | 130 + .../shuffle/streams/HadoopOffheapBuffer.java | 122 + .../HadoopEmbeddedTaskExecutor.java | 153 + .../taskexecutor/HadoopExecutorService.java | 234 + .../hadoop/taskexecutor/HadoopRunnableTask.java | 293 + .../taskexecutor/HadoopTaskExecutorAdapter.java | 59 + .../hadoop/taskexecutor/HadoopTaskState.java | 38 + .../hadoop/taskexecutor/HadoopTaskStatus.java | 116 + .../external/HadoopExternalTaskExecutor.java | 976 ++ .../external/HadoopExternalTaskMetadata.java | 67 + .../external/HadoopJobInfoUpdateRequest.java | 113 + .../external/HadoopPrepareForJobRequest.java | 130 + .../external/HadoopProcessDescriptor.java | 149 + .../external/HadoopProcessStartedAck.java | 47 + .../external/HadoopTaskExecutionRequest.java | 114 + .../external/HadoopTaskFinishedMessage.java | 94 + .../child/HadoopChildProcessRunner.java | 471 + .../child/HadoopExternalProcessStarter.java | 301 + .../HadoopAbstractCommunicationClient.java | 96 + .../HadoopCommunicationClient.java | 72 + .../HadoopExternalCommunication.java | 1460 ++ .../HadoopHandshakeTimeoutException.java | 42 + .../communication/HadoopIpcToNioAdapter.java | 248 + .../communication/HadoopMarshallerFilter.java | 86 + .../communication/HadoopMessageListener.java | 39 + .../HadoopTcpNioCommunicationClient.java | 93 + 482 files changed, 89049 insertions(+), 90134 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/ignite/blob/11b00873/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java index c909de5..dca4cf4 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java @@ -115,6 +115,7 @@ import org.apache.ignite.internal.processors.continuous.GridContinuousProcessor; import org.apache.ignite.internal.processors.datastreamer.DataStreamProcessor; import org.apache.ignite.internal.processors.datastructures.DataStructuresProcessor; import org.apache.ignite.internal.processors.hadoop.Hadoop; +import org.apache.ignite.internal.processors.hadoop.HadoopClassLoader; import org.apache.ignite.internal.processors.hadoop.HadoopProcessorAdapter; import org.apache.ignite.internal.processors.job.GridJobProcessor; import org.apache.ignite.internal.processors.jobmetrics.GridJobMetricsProcessor; @@ -1151,7 +1152,7 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable { else { HadoopProcessorAdapter cmp = null; - if (IgniteComponentType.HADOOP.inClassPath() && cfg.isPeerClassLoadingEnabled()) { + if (!ctx.hadoopHelper().isNoOp() && cfg.isPeerClassLoadingEnabled()) { U.warn(log, "Hadoop module is found in classpath, but will not be started because peer class " + "loading is enabled (set IgniteConfiguration.peerClassLoadingEnabled to \"false\" if you want " + "to use Hadoop module)."); http://git-wip-us.apache.org/repos/asf/ignite/blob/11b00873/modules/core/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopClassLoader.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopClassLoader.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopClassLoader.java index 90a2529..d2c9014 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopClassLoader.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopClassLoader.java @@ -53,11 +53,11 @@ public class HadoopClassLoader extends URLClassLoader implements ClassCache { public static final String CLS_SHUTDOWN_HOOK_MANAGER = "org.apache.hadoop.util.ShutdownHookManager"; /** Hadoop class name: Daemon replacement. */ - public static final String CLS_DAEMON_REPLACE = "org.apache.ignite.internal.processors.hadoop.v2.HadoopDaemon"; + public static final String CLS_DAEMON_REPLACE = "org.apache.ignite.internal.processors.hadoop.impl.v2.HadoopDaemon"; /** Hadoop class name: ShutdownHookManager replacement. */ public static final String CLS_SHUTDOWN_HOOK_MANAGER_REPLACE = - "org.apache.ignite.internal.processors.hadoop.v2.HadoopShutdownHookManager"; + "org.apache.ignite.internal.processors.hadoop.impl.v2.HadoopShutdownHookManager"; /** */ private static final URLClassLoader APP_CLS_LDR = (URLClassLoader)HadoopClassLoader.class.getClassLoader(); @@ -275,22 +275,16 @@ public class HadoopClassLoader extends URLClassLoader implements ClassCache { @Override protected Class<?> loadClass(String name, boolean resolve) throws ClassNotFoundException { try { // Always load Hadoop classes explicitly, since Hadoop can be available in App classpath. - if (helper.isHadoop(name)) { - if (name.equals(CLS_SHUTDOWN_HOOK_MANAGER)) // Dirty hack to get rid of Hadoop shutdown hooks. - return loadReplace(name, CLS_SHUTDOWN_HOOK_MANAGER_REPLACE); - else if (name.equals(CLS_DAEMON)) - // We replace this in order to be able to forcibly stop some daemon threads - // that otherwise never stop (e.g. PeerCache runnables): - return loadReplace(name, CLS_DAEMON_REPLACE); - - return loadClassExplicitly(name, resolve); - } + if (name.equals(CLS_SHUTDOWN_HOOK_MANAGER)) // Dirty hack to get rid of Hadoop shutdown hooks. + return loadReplace(name, CLS_SHUTDOWN_HOOK_MANAGER_REPLACE); + else if (name.equals(CLS_DAEMON)) + // We replace this in order to be able to forcibly stop some daemon threads + // that otherwise never stop (e.g. PeerCache runnables): + return loadReplace(name, CLS_DAEMON_REPLACE); // For Ignite Hadoop and IGFS classes we have to check if they depend on Hadoop. - if (helper.isHadoopIgfs(name)) { - if (hasExternalDependencies(name)) - return loadClassExplicitly(name, resolve); - } + if (loadByCurrentClassloader(name)) + return loadClassExplicitly(name, resolve); return super.loadClass(name, resolve); } @@ -343,6 +337,40 @@ public class HadoopClassLoader extends URLClassLoader implements ClassCache { } /** + * Check whether file must be loaded with current class loader, or normal delegation model should be used. + * <p> + * Override is only necessary for Ignite classes which have direct or transitive dependencies on Hadoop classes. + * These are all classes from "org.apache.ignite.internal.processors.hadoop.impl" package, + * and these are several well-know classes from "org.apache.ignite.hadoop" package. + * + * @param clsName Class name. + * @return Whether class must be loaded by current classloader without delegation. + */ + @SuppressWarnings("RedundantIfStatement") + private static boolean loadByCurrentClassloader(String clsName) { + // All impl classes. + if (clsName.startsWith("org.apache.ignite.internal.processors.hadoop.impl")) + return true; + + // Several classes from public API. + if (clsName.startsWith("org.apache.ignite.hadoop")) { + // We use "contains" instead of "equals" to handle subclasses properly. + if (clsName.contains("org.apache.ignite.hadoop.fs.v1.IgniteHadoopFileSystem") || + clsName.contains("org.apache.ignite.hadoop.fs.v2.IgniteHadoopFileSystem") || + clsName.contains("org.apache.ignite.hadoop.mapreduce.IgniteHadoopClientProtocolProvider")) + return true; + } + + // TODO: Move suites to "impl" package. + // Test suites (to be removed). + if (clsName.equals("org.apache.ignite.testsuites.IgniteHadoopTestSuite") || + clsName.equals("org.apache.ignite.testsuites.IgniteIgfsLinuxAndMacOSTestSuite")) + return true; + + return false; + } + + /** * @param name Class name. * @param resolve Resolve class. * @return Class. @@ -371,16 +399,6 @@ public class HadoopClassLoader extends URLClassLoader implements ClassCache { } /** - * Check whether class has external dependencies on Hadoop. - * - * @param clsName Class name. - * @return {@code True} if class has external dependencies. - */ - boolean hasExternalDependencies(String clsName) { - return helper.hasExternalDependencies(clsName, getParent()); - } - - /** * @param urls URLs. * @return URLs. */ http://git-wip-us.apache.org/repos/asf/ignite/blob/11b00873/modules/core/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopDefaultJobInfo.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopDefaultJobInfo.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopDefaultJobInfo.java new file mode 100644 index 0000000..ae17ac8 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopDefaultJobInfo.java @@ -0,0 +1,156 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.processors.hadoop; + +import java.io.Externalizable; +import java.io.IOException; +import java.io.ObjectInput; +import java.io.ObjectOutput; +import java.lang.reflect.Constructor; +import java.util.HashMap; +import java.util.Map; +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.IgniteLogger; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.jetbrains.annotations.Nullable; + +/** + * Hadoop job info based on default Hadoop configuration. + */ +public class HadoopDefaultJobInfo implements HadoopJobInfo, Externalizable { + /** */ + private static final long serialVersionUID = 5489900236464999951L; + + /** {@code true} If job has combiner. */ + private boolean hasCombiner; + + /** Number of reducers configured for job. */ + private int numReduces; + + /** Configuration. */ + private Map<String,String> props = new HashMap<>(); + + /** Job name. */ + private String jobName; + + /** User name. */ + private String user; + + /** + * Default constructor required by {@link Externalizable}. + */ + public HadoopDefaultJobInfo() { + // No-op. + } + + /** + * Constructor. + * + * @param jobName Job name. + * @param user User name. + * @param hasCombiner {@code true} If job has combiner. + * @param numReduces Number of reducers configured for job. + * @param props All other properties of the job. + */ + public HadoopDefaultJobInfo(String jobName, String user, boolean hasCombiner, int numReduces, + Map<String, String> props) { + this.jobName = jobName; + this.user = user; + this.hasCombiner = hasCombiner; + this.numReduces = numReduces; + this.props = props; + } + + /** {@inheritDoc} */ + @Nullable @Override public String property(String name) { + return props.get(name); + } + + /** {@inheritDoc} */ + @Override public HadoopJob createJob(Class<? extends HadoopJob> jobCls, HadoopJobId jobId, IgniteLogger log, + @Nullable String[] libNames, HadoopHelper helper) throws IgniteCheckedException { + assert jobCls != null; + + try { + Constructor<? extends HadoopJob> constructor = jobCls.getConstructor(HadoopJobId.class, + HadoopDefaultJobInfo.class, IgniteLogger.class, String[].class, HadoopHelper.class); + + return constructor.newInstance(jobId, this, log, libNames, helper); + } + catch (Throwable t) { + if (t instanceof Error) + throw (Error)t; + + throw new IgniteCheckedException(t); + } + } + + /** {@inheritDoc} */ + @Override public boolean hasCombiner() { + return hasCombiner; + } + + /** {@inheritDoc} */ + @Override public boolean hasReducer() { + return reducers() > 0; + } + + /** {@inheritDoc} */ + @Override public int reducers() { + return numReduces; + } + + /** {@inheritDoc} */ + @Override public String jobName() { + return jobName; + } + + /** {@inheritDoc} */ + @Override public String user() { + return user; + } + + /** {@inheritDoc} */ + @Override public void writeExternal(ObjectOutput out) throws IOException { + U.writeString(out, jobName); + U.writeString(out, user); + + out.writeBoolean(hasCombiner); + out.writeInt(numReduces); + + U.writeStringMap(out, props); + } + + /** {@inheritDoc} */ + @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException { + jobName = U.readString(in); + user = U.readString(in); + + hasCombiner = in.readBoolean(); + numReduces = in.readInt(); + + props = U.readStringMap(in); + } + + /** + * @return Properties of the job. + */ + public Map<String, String> properties() { + return props; + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/ignite/blob/11b00873/modules/core/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopHelper.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopHelper.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopHelper.java index 9297a2f..a8fee79 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopHelper.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopHelper.java @@ -25,6 +25,11 @@ import org.jetbrains.annotations.Nullable; */ public interface HadoopHelper { /** + * @return Whether this is no-op implementation. + */ + public boolean isNoOp(); + + /** * Get common Hadoop class loader. * * @return Common Hadoop class loader. @@ -42,32 +47,9 @@ public interface HadoopHelper { public byte[] loadReplace(InputStream in, final String originalName, final String replaceName); /** - * @param cls Class name. - * @return {@code true} If this is Hadoop class. - */ - public boolean isHadoop(String cls); - - /** - * Need to parse only Ignite Hadoop and IGFS classes. - * - * @param cls Class name. - * @return {@code true} if we need to check this class. - */ - public boolean isHadoopIgfs(String cls); - - /** * @param ldr Loader. * @param clsName Class. * @return Input stream. */ @Nullable public InputStream loadClassBytes(ClassLoader ldr, String clsName); - - /** - * Check whether class has external dependencies on Hadoop. - * - * @param clsName Class name. - * @param parentClsLdr Parent class loader. - * @return {@code True} if class has external dependencies. - */ - public boolean hasExternalDependencies(String clsName, ClassLoader parentClsLdr); } \ No newline at end of file http://git-wip-us.apache.org/repos/asf/ignite/blob/11b00873/modules/core/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopNoopHelper.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopNoopHelper.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopNoopHelper.java index 32abc91..d3348ca 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopNoopHelper.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopNoopHelper.java @@ -38,12 +38,12 @@ public class HadoopNoopHelper implements HadoopHelper { } /** {@inheritDoc} */ - @Override public HadoopClassLoader commonClassLoader() { - throw unsupported(); + @Override public boolean isNoOp() { + return true; } /** {@inheritDoc} */ - @Override public boolean hasExternalDependencies(String clsName, ClassLoader parentClsLdr) { + @Override public HadoopClassLoader commonClassLoader() { throw unsupported(); } @@ -53,16 +53,6 @@ public class HadoopNoopHelper implements HadoopHelper { } /** {@inheritDoc} */ - @Override public boolean isHadoop(String cls) { - throw unsupported(); - } - - /** {@inheritDoc} */ - @Override public boolean isHadoopIgfs(String cls) { - throw unsupported(); - } - - /** {@inheritDoc} */ @Nullable @Override public InputStream loadClassBytes(ClassLoader ldr, String clsName) { throw unsupported(); } http://git-wip-us.apache.org/repos/asf/ignite/blob/11b00873/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsImpl.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsImpl.java index 45596a3..c379c09 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsImpl.java @@ -182,8 +182,11 @@ public final class IgfsImpl implements IgfsEx { data = igfsCtx.data(); secondaryFs = cfg.getSecondaryFileSystem(); + if (secondaryFs instanceof IgfsKernalContextAware) + ((IgfsKernalContextAware)secondaryFs).setKernalContext(igfsCtx.kernalContext()); + if (secondaryFs instanceof LifecycleAware) - ((LifecycleAware) secondaryFs).start(); + ((LifecycleAware)secondaryFs).start(); /* Default IGFS mode. */ IgfsMode dfltMode; http://git-wip-us.apache.org/repos/asf/ignite/blob/11b00873/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsKernalContextAware.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsKernalContextAware.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsKernalContextAware.java new file mode 100644 index 0000000..7f59db4 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsKernalContextAware.java @@ -0,0 +1,32 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.processors.igfs; + +import org.apache.ignite.internal.GridKernalContext; + +/** + * Indicates whether particular file system accepts kernal context. + */ +public interface IgfsKernalContextAware { + /** + * Set kernal context. + * + * @param ctx Kernal context. + */ + public void setKernalContext(GridKernalContext ctx); +} http://git-wip-us.apache.org/repos/asf/ignite/blob/11b00873/modules/hadoop-impl/pom.xml ---------------------------------------------------------------------- diff --git a/modules/hadoop-impl/pom.xml b/modules/hadoop-impl/pom.xml index b82a913..faa6653 100644 --- a/modules/hadoop-impl/pom.xml +++ b/modules/hadoop-impl/pom.xml @@ -57,36 +57,42 @@ <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-annotations</artifactId> <version>${hadoop.version}</version> + <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-auth</artifactId> <version>${hadoop.version}</version> + <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-common</artifactId> <version>${hadoop.version}</version> + <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-hdfs</artifactId> <version>${hadoop.version}</version> + <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-mapreduce-client-common</artifactId> <version>${hadoop.version}</version> + <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-mapreduce-client-core</artifactId> <version>${hadoop.version}</version> + <scope>provided</scope> </dependency> <dependency> http://git-wip-us.apache.org/repos/asf/ignite/blob/11b00873/modules/hadoop-impl/src/main/java/org/apache/ignite/hadoop/fs/v1/IgniteHadoopFileSystem.java ---------------------------------------------------------------------- diff --git a/modules/hadoop-impl/src/main/java/org/apache/ignite/hadoop/fs/v1/IgniteHadoopFileSystem.java b/modules/hadoop-impl/src/main/java/org/apache/ignite/hadoop/fs/v1/IgniteHadoopFileSystem.java index 85fc76e..7133c08 100644 --- a/modules/hadoop-impl/src/main/java/org/apache/ignite/hadoop/fs/v1/IgniteHadoopFileSystem.java +++ b/modules/hadoop-impl/src/main/java/org/apache/ignite/hadoop/fs/v1/IgniteHadoopFileSystem.java @@ -44,12 +44,12 @@ import org.apache.ignite.igfs.IgfsPathSummary; import org.apache.ignite.internal.igfs.common.IgfsLogger; import org.apache.ignite.internal.processors.hadoop.delegate.HadoopDelegateUtils; import org.apache.ignite.internal.processors.hadoop.delegate.HadoopFileSystemFactoryDelegate; -import org.apache.ignite.internal.processors.hadoop.igfs.HadoopIgfsInputStream; -import org.apache.ignite.internal.processors.hadoop.igfs.HadoopIgfsOutputStream; -import org.apache.ignite.internal.processors.hadoop.igfs.HadoopIgfsProxyInputStream; -import org.apache.ignite.internal.processors.hadoop.igfs.HadoopIgfsProxyOutputStream; -import org.apache.ignite.internal.processors.hadoop.igfs.HadoopIgfsStreamDelegate; -import org.apache.ignite.internal.processors.hadoop.igfs.HadoopIgfsWrapper; +import org.apache.ignite.internal.processors.hadoop.impl.igfs.HadoopIgfsInputStream; +import org.apache.ignite.internal.processors.hadoop.impl.igfs.HadoopIgfsOutputStream; +import org.apache.ignite.internal.processors.hadoop.impl.igfs.HadoopIgfsProxyInputStream; +import org.apache.ignite.internal.processors.hadoop.impl.igfs.HadoopIgfsProxyOutputStream; +import org.apache.ignite.internal.processors.hadoop.impl.igfs.HadoopIgfsStreamDelegate; +import org.apache.ignite.internal.processors.hadoop.impl.igfs.HadoopIgfsWrapper; import org.apache.ignite.internal.processors.igfs.IgfsHandshakeResponse; import org.apache.ignite.internal.processors.igfs.IgfsModeResolver; import org.apache.ignite.internal.processors.igfs.IgfsPaths; @@ -79,13 +79,13 @@ import java.util.concurrent.atomic.AtomicBoolean; import static org.apache.ignite.configuration.FileSystemConfiguration.DFLT_IGFS_LOG_BATCH_SIZE; import static org.apache.ignite.configuration.FileSystemConfiguration.DFLT_IGFS_LOG_DIR; import static org.apache.ignite.igfs.IgfsMode.PROXY; -import static org.apache.ignite.internal.processors.hadoop.fs.HadoopParameters.PARAM_IGFS_COLOCATED_WRITES; -import static org.apache.ignite.internal.processors.hadoop.fs.HadoopParameters.PARAM_IGFS_LOG_BATCH_SIZE; -import static org.apache.ignite.internal.processors.hadoop.fs.HadoopParameters.PARAM_IGFS_LOG_DIR; -import static org.apache.ignite.internal.processors.hadoop.fs.HadoopParameters.PARAM_IGFS_LOG_ENABLED; -import static org.apache.ignite.internal.processors.hadoop.fs.HadoopParameters.PARAM_IGFS_PREFER_LOCAL_WRITES; -import static org.apache.ignite.internal.processors.hadoop.fs.HadoopParameters.PARAM_IGFS_SEQ_READS_BEFORE_PREFETCH; -import static org.apache.ignite.internal.processors.hadoop.igfs.HadoopIgfsUtils.parameter; +import static org.apache.ignite.internal.processors.hadoop.impl.fs.HadoopParameters.PARAM_IGFS_COLOCATED_WRITES; +import static org.apache.ignite.internal.processors.hadoop.impl.fs.HadoopParameters.PARAM_IGFS_LOG_BATCH_SIZE; +import static org.apache.ignite.internal.processors.hadoop.impl.fs.HadoopParameters.PARAM_IGFS_LOG_DIR; +import static org.apache.ignite.internal.processors.hadoop.impl.fs.HadoopParameters.PARAM_IGFS_LOG_ENABLED; +import static org.apache.ignite.internal.processors.hadoop.impl.fs.HadoopParameters.PARAM_IGFS_PREFER_LOCAL_WRITES; +import static org.apache.ignite.internal.processors.hadoop.impl.fs.HadoopParameters.PARAM_IGFS_SEQ_READS_BEFORE_PREFETCH; +import static org.apache.ignite.internal.processors.hadoop.impl.igfs.HadoopIgfsUtils.parameter; import static org.apache.ignite.internal.processors.igfs.IgfsEx.IGFS_SCHEME; /** http://git-wip-us.apache.org/repos/asf/ignite/blob/11b00873/modules/hadoop-impl/src/main/java/org/apache/ignite/hadoop/fs/v2/IgniteHadoopFileSystem.java ---------------------------------------------------------------------- diff --git a/modules/hadoop-impl/src/main/java/org/apache/ignite/hadoop/fs/v2/IgniteHadoopFileSystem.java b/modules/hadoop-impl/src/main/java/org/apache/ignite/hadoop/fs/v2/IgniteHadoopFileSystem.java index 32e51df..18b8bf9 100644 --- a/modules/hadoop-impl/src/main/java/org/apache/ignite/hadoop/fs/v2/IgniteHadoopFileSystem.java +++ b/modules/hadoop-impl/src/main/java/org/apache/ignite/hadoop/fs/v2/IgniteHadoopFileSystem.java @@ -49,12 +49,12 @@ import org.apache.ignite.internal.igfs.common.IgfsLogger; import org.apache.ignite.internal.processors.hadoop.delegate.HadoopDelegateUtils; import org.apache.ignite.internal.processors.hadoop.delegate.HadoopFileSystemFactoryDelegate; import org.apache.ignite.internal.processors.hadoop.igfs.HadoopIgfsEndpoint; -import org.apache.ignite.internal.processors.hadoop.igfs.HadoopIgfsInputStream; -import org.apache.ignite.internal.processors.hadoop.igfs.HadoopIgfsOutputStream; -import org.apache.ignite.internal.processors.hadoop.igfs.HadoopIgfsProxyInputStream; -import org.apache.ignite.internal.processors.hadoop.igfs.HadoopIgfsProxyOutputStream; -import org.apache.ignite.internal.processors.hadoop.igfs.HadoopIgfsStreamDelegate; -import org.apache.ignite.internal.processors.hadoop.igfs.HadoopIgfsWrapper; +import org.apache.ignite.internal.processors.hadoop.impl.igfs.HadoopIgfsInputStream; +import org.apache.ignite.internal.processors.hadoop.impl.igfs.HadoopIgfsOutputStream; +import org.apache.ignite.internal.processors.hadoop.impl.igfs.HadoopIgfsProxyInputStream; +import org.apache.ignite.internal.processors.hadoop.impl.igfs.HadoopIgfsProxyOutputStream; +import org.apache.ignite.internal.processors.hadoop.impl.igfs.HadoopIgfsStreamDelegate; +import org.apache.ignite.internal.processors.hadoop.impl.igfs.HadoopIgfsWrapper; import org.apache.ignite.internal.processors.igfs.IgfsHandshakeResponse; import org.apache.ignite.internal.processors.igfs.IgfsModeResolver; import org.apache.ignite.internal.processors.igfs.IgfsPaths; @@ -87,13 +87,13 @@ import static org.apache.ignite.configuration.FileSystemConfiguration.DFLT_IGFS_ import static org.apache.ignite.configuration.FileSystemConfiguration.DFLT_IGFS_LOG_DIR; import static org.apache.ignite.hadoop.fs.v1.IgniteHadoopFileSystem.getFsHadoopUser; import static org.apache.ignite.igfs.IgfsMode.PROXY; -import static org.apache.ignite.internal.processors.hadoop.fs.HadoopParameters.PARAM_IGFS_COLOCATED_WRITES; -import static org.apache.ignite.internal.processors.hadoop.fs.HadoopParameters.PARAM_IGFS_LOG_BATCH_SIZE; -import static org.apache.ignite.internal.processors.hadoop.fs.HadoopParameters.PARAM_IGFS_LOG_DIR; -import static org.apache.ignite.internal.processors.hadoop.fs.HadoopParameters.PARAM_IGFS_LOG_ENABLED; -import static org.apache.ignite.internal.processors.hadoop.fs.HadoopParameters.PARAM_IGFS_PREFER_LOCAL_WRITES; -import static org.apache.ignite.internal.processors.hadoop.fs.HadoopParameters.PARAM_IGFS_SEQ_READS_BEFORE_PREFETCH; -import static org.apache.ignite.internal.processors.hadoop.igfs.HadoopIgfsUtils.parameter; +import static org.apache.ignite.internal.processors.hadoop.impl.fs.HadoopParameters.PARAM_IGFS_COLOCATED_WRITES; +import static org.apache.ignite.internal.processors.hadoop.impl.fs.HadoopParameters.PARAM_IGFS_LOG_BATCH_SIZE; +import static org.apache.ignite.internal.processors.hadoop.impl.fs.HadoopParameters.PARAM_IGFS_LOG_DIR; +import static org.apache.ignite.internal.processors.hadoop.impl.fs.HadoopParameters.PARAM_IGFS_LOG_ENABLED; +import static org.apache.ignite.internal.processors.hadoop.impl.fs.HadoopParameters.PARAM_IGFS_PREFER_LOCAL_WRITES; +import static org.apache.ignite.internal.processors.hadoop.impl.fs.HadoopParameters.PARAM_IGFS_SEQ_READS_BEFORE_PREFETCH; +import static org.apache.ignite.internal.processors.hadoop.impl.igfs.HadoopIgfsUtils.parameter; import static org.apache.ignite.internal.processors.igfs.IgfsEx.IGFS_SCHEME; /** http://git-wip-us.apache.org/repos/asf/ignite/blob/11b00873/modules/hadoop-impl/src/main/java/org/apache/ignite/hadoop/mapreduce/IgniteHadoopClientProtocolProvider.java ---------------------------------------------------------------------- diff --git a/modules/hadoop-impl/src/main/java/org/apache/ignite/hadoop/mapreduce/IgniteHadoopClientProtocolProvider.java b/modules/hadoop-impl/src/main/java/org/apache/ignite/hadoop/mapreduce/IgniteHadoopClientProtocolProvider.java index 583af35..c806cb1 100644 --- a/modules/hadoop-impl/src/main/java/org/apache/ignite/hadoop/mapreduce/IgniteHadoopClientProtocolProvider.java +++ b/modules/hadoop-impl/src/main/java/org/apache/ignite/hadoop/mapreduce/IgniteHadoopClientProtocolProvider.java @@ -32,7 +32,7 @@ import org.apache.ignite.internal.client.GridClientConfiguration; import org.apache.ignite.internal.client.GridClientException; import org.apache.ignite.internal.client.GridClientFactory; import org.apache.ignite.internal.client.marshaller.jdk.GridClientJdkMarshaller; -import org.apache.ignite.internal.processors.hadoop.proto.HadoopClientProtocol; +import org.apache.ignite.internal.processors.hadoop.impl.proto.HadoopClientProtocol; import org.apache.ignite.internal.util.future.GridFutureAdapter; import org.apache.ignite.internal.util.typedef.F; http://git-wip-us.apache.org/repos/asf/ignite/blob/11b00873/modules/hadoop-impl/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopAttributes.java ---------------------------------------------------------------------- diff --git a/modules/hadoop-impl/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopAttributes.java b/modules/hadoop-impl/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopAttributes.java deleted file mode 100644 index 23eaa18..0000000 --- a/modules/hadoop-impl/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopAttributes.java +++ /dev/null @@ -1,168 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.ignite.internal.processors.hadoop; - -import org.apache.ignite.cluster.ClusterNode; -import org.apache.ignite.configuration.HadoopConfiguration; -import org.apache.ignite.internal.IgniteNodeAttributes; -import org.apache.ignite.internal.util.tostring.GridToStringExclude; -import org.apache.ignite.internal.util.typedef.internal.S; -import org.jetbrains.annotations.Nullable; - -import java.io.Externalizable; -import java.io.IOException; -import java.io.ObjectInput; -import java.io.ObjectOutput; -import java.util.Arrays; - -/** - * Hadoop attributes. - */ -public class HadoopAttributes implements Externalizable { - /** */ - private static final long serialVersionUID = 0L; - - /** Attribute name. */ - public static final String NAME = IgniteNodeAttributes.ATTR_PREFIX + ".hadoop"; - - /** Map-reduce planner class name. */ - private String plannerCls; - - /** External executor flag. */ - private boolean extExec; - - /** Maximum parallel tasks. */ - private int maxParallelTasks; - - /** Maximum task queue size. */ - private int maxTaskQueueSize; - - /** Library names. */ - @GridToStringExclude - private String[] libNames; - - /** Number of cores. */ - private int cores; - - /** - * Get attributes for node (if any). - * - * @param node Node. - * @return Attributes or {@code null} if Hadoop Accelerator is not enabled for node. - */ - @Nullable public static HadoopAttributes forNode(ClusterNode node) { - return node.attribute(NAME); - } - - /** - * {@link Externalizable} support. - */ - public HadoopAttributes() { - // No-op. - } - - /** - * Constructor. - * - * @param cfg Configuration. - */ - public HadoopAttributes(HadoopConfiguration cfg) { - assert cfg != null; - assert cfg.getMapReducePlanner() != null; - - plannerCls = cfg.getMapReducePlanner().getClass().getName(); - - // TODO: IGNITE-404: Get from configuration when fixed. - extExec = false; - - maxParallelTasks = cfg.getMaxParallelTasks(); - maxTaskQueueSize = cfg.getMaxTaskQueueSize(); - libNames = cfg.getNativeLibraryNames(); - - // Cores count already passed in other attributes, we add it here for convenience. - cores = Runtime.getRuntime().availableProcessors(); - } - - /** - * @return Map reduce planner class name. - */ - public String plannerClassName() { - return plannerCls; - } - - /** - * @return External execution flag. - */ - public boolean externalExecution() { - return extExec; - } - - /** - * @return Maximum parallel tasks. - */ - public int maxParallelTasks() { - return maxParallelTasks; - } - - /** - * @return Maximum task queue size. - */ - public int maxTaskQueueSize() { - return maxTaskQueueSize; - } - - - /** - * @return Native library names. - */ - public String[] nativeLibraryNames() { - return libNames; - } - - /** - * @return Number of cores on machine. - */ - public int cores() { - return cores; - } - - /** {@inheritDoc} */ - @Override public void writeExternal(ObjectOutput out) throws IOException { - out.writeObject(plannerCls); - out.writeBoolean(extExec); - out.writeInt(maxParallelTasks); - out.writeInt(maxTaskQueueSize); - out.writeObject(libNames); - out.writeInt(cores); - } - - /** {@inheritDoc} */ - @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException { - plannerCls = (String)in.readObject(); - extExec = in.readBoolean(); - maxParallelTasks = in.readInt(); - maxTaskQueueSize = in.readInt(); - libNames = (String[])in.readObject(); - cores = in.readInt(); - } - - /** {@inheritDoc} */ - @Override public String toString() { - return S.toString(HadoopAttributes.class, this, "libNames", Arrays.toString(libNames)); - } -} http://git-wip-us.apache.org/repos/asf/ignite/blob/11b00873/modules/hadoop-impl/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopComponent.java ---------------------------------------------------------------------- diff --git a/modules/hadoop-impl/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopComponent.java b/modules/hadoop-impl/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopComponent.java deleted file mode 100644 index aeda5c0..0000000 --- a/modules/hadoop-impl/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopComponent.java +++ /dev/null @@ -1,62 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.ignite.internal.processors.hadoop; - -import org.apache.ignite.IgniteCheckedException; -import org.apache.ignite.IgniteLogger; - -/** - * Abstract class for all hadoop components. - */ -public abstract class HadoopComponent { - /** Hadoop context. */ - protected HadoopContext ctx; - - /** Logger. */ - protected IgniteLogger log; - - /** - * @param ctx Hadoop context. - */ - public void start(HadoopContext ctx) throws IgniteCheckedException { - this.ctx = ctx; - - log = ctx.kernalContext().log(getClass()); - } - - /** - * Stops manager. - */ - public void stop(boolean cancel) { - // No-op. - } - - /** - * Callback invoked when all grid components are started. - */ - public void onKernalStart() throws IgniteCheckedException { - // No-op. - } - - /** - * Callback invoked before all grid components are stopped. - */ - public void onKernalStop(boolean cancel) { - // No-op. - } -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/ignite/blob/11b00873/modules/hadoop-impl/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopContext.java ---------------------------------------------------------------------- diff --git a/modules/hadoop-impl/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopContext.java b/modules/hadoop-impl/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopContext.java deleted file mode 100644 index 42a3d72..0000000 --- a/modules/hadoop-impl/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopContext.java +++ /dev/null @@ -1,201 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.ignite.internal.processors.hadoop; - -import java.util.ArrayList; -import java.util.Collection; -import java.util.List; -import java.util.UUID; -import org.apache.ignite.cluster.ClusterNode; -import org.apache.ignite.configuration.HadoopConfiguration; -import org.apache.ignite.internal.GridKernalContext; -import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; -import org.apache.ignite.internal.processors.hadoop.jobtracker.HadoopJobMetadata; -import org.apache.ignite.internal.processors.hadoop.jobtracker.HadoopJobTracker; -import org.apache.ignite.internal.processors.hadoop.shuffle.HadoopShuffle; -import org.apache.ignite.internal.processors.hadoop.taskexecutor.HadoopTaskExecutorAdapter; -import org.apache.ignite.internal.util.typedef.internal.CU; - -/** - * Hadoop accelerator context. - */ -public class HadoopContext { - /** Kernal context. */ - private GridKernalContext ctx; - - /** Hadoop configuration. */ - private HadoopConfiguration cfg; - - /** Job tracker. */ - private HadoopJobTracker jobTracker; - - /** External task executor. */ - private HadoopTaskExecutorAdapter taskExecutor; - - /** */ - private HadoopShuffle shuffle; - - /** Managers list. */ - private List<HadoopComponent> components = new ArrayList<>(); - - /** - * @param ctx Kernal context. - */ - public HadoopContext( - GridKernalContext ctx, - HadoopConfiguration cfg, - HadoopJobTracker jobTracker, - HadoopTaskExecutorAdapter taskExecutor, - HadoopShuffle shuffle - ) { - this.ctx = ctx; - this.cfg = cfg; - - this.jobTracker = add(jobTracker); - this.taskExecutor = add(taskExecutor); - this.shuffle = add(shuffle); - } - - /** - * Gets list of managers. - * - * @return List of managers. - */ - public List<HadoopComponent> components() { - return components; - } - - /** - * Gets kernal context. - * - * @return Grid kernal context instance. - */ - public GridKernalContext kernalContext() { - return ctx; - } - - /** - * Gets Hadoop configuration. - * - * @return Hadoop configuration. - */ - public HadoopConfiguration configuration() { - return cfg; - } - - /** - * Gets local node ID. Shortcut for {@code kernalContext().localNodeId()}. - * - * @return Local node ID. - */ - public UUID localNodeId() { - return ctx.localNodeId(); - } - - /** - * Gets local node order. - * - * @return Local node order. - */ - public long localNodeOrder() { - assert ctx.discovery() != null; - - return ctx.discovery().localNode().order(); - } - - /** - * @return Hadoop-enabled nodes. - */ - public Collection<ClusterNode> nodes() { - return ctx.discovery().cacheNodes(CU.SYS_CACHE_HADOOP_MR, ctx.discovery().topologyVersionEx()); - } - - /** - * @return {@code True} if - */ - public boolean jobUpdateLeader() { - long minOrder = Long.MAX_VALUE; - ClusterNode minOrderNode = null; - - for (ClusterNode node : nodes()) { - if (node.order() < minOrder) { - minOrder = node.order(); - minOrderNode = node; - } - } - - assert minOrderNode != null; - - return localNodeId().equals(minOrderNode.id()); - } - - /** - * @param meta Job metadata. - * @return {@code true} If local node is participating in job execution. - */ - public boolean isParticipating(HadoopJobMetadata meta) { - UUID locNodeId = localNodeId(); - - if (locNodeId.equals(meta.submitNodeId())) - return true; - - HadoopMapReducePlan plan = meta.mapReducePlan(); - - return plan.mapperNodeIds().contains(locNodeId) || plan.reducerNodeIds().contains(locNodeId) || jobUpdateLeader(); - } - - /** - * @return Jon tracker instance. - */ - public HadoopJobTracker jobTracker() { - return jobTracker; - } - - /** - * @return Task executor. - */ - public HadoopTaskExecutorAdapter taskExecutor() { - return taskExecutor; - } - - /** - * @return Shuffle. - */ - public HadoopShuffle shuffle() { - return shuffle; - } - - /** - * @return Map-reduce planner. - */ - public HadoopMapReducePlanner planner() { - return cfg.getMapReducePlanner(); - } - - /** - * Adds component. - * - * @param c Component to add. - * @return Added manager. - */ - private <C extends HadoopComponent> C add(C c) { - components.add(c); - - return c; - } -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/ignite/blob/11b00873/modules/hadoop-impl/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopDefaultJobInfo.java ---------------------------------------------------------------------- diff --git a/modules/hadoop-impl/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopDefaultJobInfo.java b/modules/hadoop-impl/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopDefaultJobInfo.java deleted file mode 100644 index ae17ac8..0000000 --- a/modules/hadoop-impl/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopDefaultJobInfo.java +++ /dev/null @@ -1,156 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.ignite.internal.processors.hadoop; - -import java.io.Externalizable; -import java.io.IOException; -import java.io.ObjectInput; -import java.io.ObjectOutput; -import java.lang.reflect.Constructor; -import java.util.HashMap; -import java.util.Map; -import org.apache.ignite.IgniteCheckedException; -import org.apache.ignite.IgniteLogger; -import org.apache.ignite.internal.util.typedef.internal.U; -import org.jetbrains.annotations.Nullable; - -/** - * Hadoop job info based on default Hadoop configuration. - */ -public class HadoopDefaultJobInfo implements HadoopJobInfo, Externalizable { - /** */ - private static final long serialVersionUID = 5489900236464999951L; - - /** {@code true} If job has combiner. */ - private boolean hasCombiner; - - /** Number of reducers configured for job. */ - private int numReduces; - - /** Configuration. */ - private Map<String,String> props = new HashMap<>(); - - /** Job name. */ - private String jobName; - - /** User name. */ - private String user; - - /** - * Default constructor required by {@link Externalizable}. - */ - public HadoopDefaultJobInfo() { - // No-op. - } - - /** - * Constructor. - * - * @param jobName Job name. - * @param user User name. - * @param hasCombiner {@code true} If job has combiner. - * @param numReduces Number of reducers configured for job. - * @param props All other properties of the job. - */ - public HadoopDefaultJobInfo(String jobName, String user, boolean hasCombiner, int numReduces, - Map<String, String> props) { - this.jobName = jobName; - this.user = user; - this.hasCombiner = hasCombiner; - this.numReduces = numReduces; - this.props = props; - } - - /** {@inheritDoc} */ - @Nullable @Override public String property(String name) { - return props.get(name); - } - - /** {@inheritDoc} */ - @Override public HadoopJob createJob(Class<? extends HadoopJob> jobCls, HadoopJobId jobId, IgniteLogger log, - @Nullable String[] libNames, HadoopHelper helper) throws IgniteCheckedException { - assert jobCls != null; - - try { - Constructor<? extends HadoopJob> constructor = jobCls.getConstructor(HadoopJobId.class, - HadoopDefaultJobInfo.class, IgniteLogger.class, String[].class, HadoopHelper.class); - - return constructor.newInstance(jobId, this, log, libNames, helper); - } - catch (Throwable t) { - if (t instanceof Error) - throw (Error)t; - - throw new IgniteCheckedException(t); - } - } - - /** {@inheritDoc} */ - @Override public boolean hasCombiner() { - return hasCombiner; - } - - /** {@inheritDoc} */ - @Override public boolean hasReducer() { - return reducers() > 0; - } - - /** {@inheritDoc} */ - @Override public int reducers() { - return numReduces; - } - - /** {@inheritDoc} */ - @Override public String jobName() { - return jobName; - } - - /** {@inheritDoc} */ - @Override public String user() { - return user; - } - - /** {@inheritDoc} */ - @Override public void writeExternal(ObjectOutput out) throws IOException { - U.writeString(out, jobName); - U.writeString(out, user); - - out.writeBoolean(hasCombiner); - out.writeInt(numReduces); - - U.writeStringMap(out, props); - } - - /** {@inheritDoc} */ - @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException { - jobName = U.readString(in); - user = U.readString(in); - - hasCombiner = in.readBoolean(); - numReduces = in.readInt(); - - props = U.readStringMap(in); - } - - /** - * @return Properties of the job. - */ - public Map<String, String> properties() { - return props; - } -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/ignite/blob/11b00873/modules/hadoop-impl/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopImpl.java ---------------------------------------------------------------------- diff --git a/modules/hadoop-impl/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopImpl.java b/modules/hadoop-impl/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopImpl.java deleted file mode 100644 index ed2657e..0000000 --- a/modules/hadoop-impl/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopImpl.java +++ /dev/null @@ -1,134 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.ignite.internal.processors.hadoop; - -import org.apache.ignite.IgniteCheckedException; -import org.apache.ignite.configuration.HadoopConfiguration; -import org.apache.ignite.internal.IgniteInternalFuture; -import org.apache.ignite.internal.processors.hadoop.counter.HadoopCounters; -import org.apache.ignite.internal.util.GridSpinBusyLock; -import org.jetbrains.annotations.Nullable; - -/** - * Hadoop facade implementation. - */ -public class HadoopImpl implements Hadoop { - /** Hadoop processor. */ - private final HadoopProcessor proc; - - /** Busy lock. */ - private final GridSpinBusyLock busyLock = new GridSpinBusyLock(); - - /** - * Constructor. - * - * @param proc Hadoop processor. - */ - HadoopImpl(HadoopProcessor proc) { - this.proc = proc; - } - - /** {@inheritDoc} */ - @Override public HadoopConfiguration configuration() { - return proc.config(); - } - - /** {@inheritDoc} */ - @Override public HadoopJobId nextJobId() { - if (busyLock.enterBusy()) { - try { - return proc.nextJobId(); - } - finally { - busyLock.leaveBusy(); - } - } - else - throw new IllegalStateException("Failed to get next job ID (grid is stopping)."); - } - - /** {@inheritDoc} */ - @Override public IgniteInternalFuture<?> submit(HadoopJobId jobId, HadoopJobInfo jobInfo) { - if (busyLock.enterBusy()) { - try { - return proc.submit(jobId, jobInfo); - } - finally { - busyLock.leaveBusy(); - } - } - else - throw new IllegalStateException("Failed to submit job (grid is stopping)."); - } - - /** {@inheritDoc} */ - @Nullable @Override public HadoopJobStatus status(HadoopJobId jobId) throws IgniteCheckedException { - if (busyLock.enterBusy()) { - try { - return proc.status(jobId); - } - finally { - busyLock.leaveBusy(); - } - } - else - throw new IllegalStateException("Failed to get job status (grid is stopping)."); - } - - /** {@inheritDoc} */ - @Nullable @Override public HadoopCounters counters(HadoopJobId jobId) throws IgniteCheckedException { - if (busyLock.enterBusy()) { - try { - return proc.counters(jobId); - } - finally { - busyLock.leaveBusy(); - } - } - else - throw new IllegalStateException("Failed to get job counters (grid is stopping)."); - } - - /** {@inheritDoc} */ - @Nullable @Override public IgniteInternalFuture<?> finishFuture(HadoopJobId jobId) throws IgniteCheckedException { - if (busyLock.enterBusy()) { - try { - return proc.finishFuture(jobId); - } - finally { - busyLock.leaveBusy(); - } - } - else - throw new IllegalStateException("Failed to get job finish future (grid is stopping)."); - } - - /** {@inheritDoc} */ - @Override public boolean kill(HadoopJobId jobId) throws IgniteCheckedException { - if (busyLock.enterBusy()) { - try { - return proc.kill(jobId); - } - finally { - busyLock.leaveBusy(); - } - } - else - throw new IllegalStateException("Failed to kill job (grid is stopping)."); - } -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/ignite/blob/11b00873/modules/hadoop-impl/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopMapReduceCounterGroup.java ---------------------------------------------------------------------- diff --git a/modules/hadoop-impl/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopMapReduceCounterGroup.java b/modules/hadoop-impl/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopMapReduceCounterGroup.java deleted file mode 100644 index 4e03e17..0000000 --- a/modules/hadoop-impl/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopMapReduceCounterGroup.java +++ /dev/null @@ -1,123 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.ignite.internal.processors.hadoop; - -import java.io.DataInput; -import java.io.DataOutput; -import java.io.IOException; -import java.util.Iterator; -import org.apache.hadoop.mapreduce.Counter; -import org.apache.hadoop.mapreduce.CounterGroup; -import org.apache.hadoop.mapreduce.counters.CounterGroupBase; - -/** - * Hadoop +counter group adapter. - */ -class HadoopMapReduceCounterGroup implements CounterGroup { - /** Counters. */ - private final HadoopMapReduceCounters cntrs; - - /** Group name. */ - private final String name; - - /** - * Creates new instance. - * - * @param cntrs Client counters instance. - * @param name Group name. - */ - HadoopMapReduceCounterGroup(HadoopMapReduceCounters cntrs, String name) { - this.cntrs = cntrs; - this.name = name; - } - - /** {@inheritDoc} */ - @Override public String getName() { - return name; - } - - /** {@inheritDoc} */ - @Override public String getDisplayName() { - return name; - } - - /** {@inheritDoc} */ - @Override public void setDisplayName(String displayName) { - // No-op. - } - - /** {@inheritDoc} */ - @Override public void addCounter(Counter counter) { - addCounter(counter.getName(), counter.getDisplayName(), 0); - } - - /** {@inheritDoc} */ - @Override public Counter addCounter(String name, String displayName, long value) { - final Counter counter = cntrs.findCounter(this.name, name); - - counter.setValue(value); - - return counter; - } - - /** {@inheritDoc} */ - @Override public Counter findCounter(String counterName, String displayName) { - return cntrs.findCounter(name, counterName); - } - - /** {@inheritDoc} */ - @Override public Counter findCounter(String counterName, boolean create) { - return cntrs.findCounter(name, counterName, create); - } - - /** {@inheritDoc} */ - @Override public Counter findCounter(String counterName) { - return cntrs.findCounter(name, counterName); - } - - /** {@inheritDoc} */ - @Override public int size() { - return cntrs.groupSize(name); - } - - /** {@inheritDoc} */ - @Override public void incrAllCounters(CounterGroupBase<Counter> rightGroup) { - for (final Counter counter : rightGroup) - cntrs.findCounter(name, counter.getName()).increment(counter.getValue()); - } - - /** {@inheritDoc} */ - @Override public CounterGroupBase<Counter> getUnderlyingGroup() { - return this; - } - - /** {@inheritDoc} */ - @Override public Iterator<Counter> iterator() { - return cntrs.iterateGroup(name); - } - - /** {@inheritDoc} */ - @Override public void write(DataOutput out) throws IOException { - throw new UnsupportedOperationException("not implemented"); - } - - /** {@inheritDoc} */ - @Override public void readFields(DataInput in) throws IOException { - throw new UnsupportedOperationException("not implemented"); - } -} \ No newline at end of file
