1 /*
2 * Licensed to the Apache Software Foundation (ASF) under one
3 * or more contributor license agreements. See the NOTICE file
4 * distributed with this work for additional information
5 * regarding copyright ownership. The ASF licenses this file
6 * to you under the Apache License, Version 2.0 (the
7 * "License"); you may not use this file except in compliance
8 * with the License. You may obtain a copy of the License at
9 *
10 * http://www.apache.org/licenses/LICENSE-2.0
11 *
12 * Unless required by applicable law or agreed to in writing, software
13 * distributed under the License is distributed on an "AS IS" BASIS,
14 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15 * See the License for the specific language governing permissions and
16 * limitations under the License.
17 */
18
19 package org.apache.hadoop.chukwa.extraction.demux.processor.reducer;
20
21
22 import java.util.HashMap;
23 import org.apache.log4j.Logger;
24
25 public class ReduceProcessorFactory {
26 static Logger log = Logger.getLogger(ReduceProcessorFactory.class);
27
28 // TODO
29 // add new mapper package at the end.
30 // We should have a more generic way to do this.
31 // Ex: read from config
32 // list of alias
33 // and
34 // alias -> processor class
35
36 // ******** WARNING ********
37 // If the ReduceProcessor is not there use Identity instead
38
39 private static HashMap<String, ReduceProcessor> processors = new HashMap<String, ReduceProcessor>(); // registry
40
41 private ReduceProcessorFactory() {
42 }
43
44 public static ReduceProcessor getProcessor(String reduceType, String defaultProcessor)
45 throws UnknownReduceTypeException {
46 String path = "org.apache.hadoop.chukwa.extraction.demux.processor.reducer."
47 + reduceType;
48 if (processors.containsKey(reduceType)) {
49 return processors.get(reduceType);
50 } else {
51 ReduceProcessor processor = null;
52 try {
53 processor = (ReduceProcessor) Class.forName(path).getConstructor()
54 .newInstance();
55 } catch (ClassNotFoundException e) {
56 // ******** WARNING ********
57 // If the ReduceProcessor is not there see if there is a configured
58 // default processor. If not, fall back on the Identity instead
59 if(defaultProcessor != null) {
60 processor = getProcessor(defaultProcessor, null);
61 }
62 else {
63 processor = getProcessor("IdentityReducer", null);
64 }
65 register(reduceType, processor);
66 return processor;
67 } catch (Exception e) {
68 throw new UnknownReduceTypeException("error constructing processor", e);
69 }
70
71 // TODO using a ThreadSafe/reuse flag to actually decide if we want
72 // to reuse the same processor again and again
73 register(reduceType, processor);
74 return processor;
75 }
76 }
77
78 /**
79 * Register a specific parser for a {@link ReduceProcessor} implementation.
80 */
81 public static synchronized void register(String reduceType,
82 ReduceProcessor processor) {
83 log.info("register " + processor.getClass().getName()
84 + " for this recordType :" + reduceType);
85 if (processors.containsKey(reduceType)) {
86 throw new DuplicateReduceProcessorException(
87 "Duplicate processor for recordType:" + reduceType);
88 }
89 ReduceProcessorFactory.processors.put(reduceType, processor);
90 }
91
92 }