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
24 import org.apache.log4j.Logger;
25
26 public class ReduceProcessorFactory {
27 static Logger log = Logger.getLogger(ReduceProcessorFactory.class);
28
29 // TODO
30 // add new mapper package at the end.
31 // We should have a more generic way to do this.
32 // Ex: read from config
33 // list of alias
34 // and
35 // alias -> processor class
36
37 // ******** WARNING ********
38 // If the ReduceProcessor is not there use Identity instead
39
40 private static HashMap<String, ReduceProcessor> processors = new HashMap<String, ReduceProcessor>(); // registry
41
42 private ReduceProcessorFactory() {
43 }
44
45 /**
46 * Register a specific parser for a {@link ReduceProcessor} implementation.
47 * @param reduceType
48 * @param processor
49 */
50 public static synchronized void register(String reduceType,
51 ReduceProcessor processor) {
52 log.info("register " + processor.getClass().getName()
53 + " for this recordType :" + reduceType);
54 if (processors.containsKey(reduceType)) {
55 throw new DuplicateReduceProcessorException(
56 "Duplicate processor for recordType:" + reduceType);
57 }
58 ReduceProcessorFactory.processors.put(reduceType, processor);
59 }
60
61 public static ReduceProcessor getProcessor(String processorClass) throws UnknownReduceTypeException {
62 if (processors.containsKey(processorClass)) {
63 return processors.get(processorClass);
64 } else {
65 ReduceProcessor processor = null;
66 try {
67 processor = (ReduceProcessor) Class.forName(processorClass).newInstance();
68 } catch (ClassNotFoundException e) {
69 throw new UnknownReduceTypeException("Unknown reducer class for:" + processorClass, e);
70 } catch (Exception e) {
71 throw new UnknownReduceTypeException("error constructing processor", e);
72 }
73 // TODO using a ThreadSafe/reuse flag to actually decide if we want
74 // to reuse the same processor again and again
75 register(processorClass, processor);
76 return processor;
77 }
78
79 }
80
81
82 }