GitHub PullRequestReviewCommentEvent at 2014-05-22 04:29:13
Repository: spring-hadoop
Actor: markpollack
Payload: {"comment":{"url":"https://api.github.com/repos/spring-projects/spring-hadoop/pulls/comments/12934603","id":12934603,"diff_hunk":"@@ -0,0 +1,299 @@\n+/*\n+ * Copyright 2014 the original author or authors.\n+ *\n+ * Licensed under the Apache License, Version 2.0 (the \"License\");\n+ * you may not use this file except in compliance with the License.\n+ * You may obtain a copy of the License at\n+ *\n+ *      http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ * Unless required by applicable law or agreed to in writing, software\n+ * distributed under the License is distributed on an \"AS IS\" BASIS,\n+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n+ * See the License for the specific language governing permissions and\n+ * limitations under the License.\n+ */\n+package org.springframework.data.hadoop.store.output;\n+\n+import java.io.IOException;\n+import java.util.Map;\n+import java.util.concurrent.ConcurrentHashMap;\n+\n+import org.apache.commons.logging.Log;\n+import org.apache.commons.logging.LogFactory;\n+import org.apache.hadoop.conf.Configuration;\n+import org.apache.hadoop.fs.Path;\n+import org.springframework.data.hadoop.store.DataStoreWriter;\n+import org.springframework.data.hadoop.store.PartitionDataStoreWriter;\n+import org.springframework.data.hadoop.store.codec.CodecInfo;\n+import org.springframework.data.hadoop.store.partition.PartitionStrategy;\n+import org.springframework.data.hadoop.store.strategy.naming.FileNamingStrategy;\n+import org.springframework.data.hadoop.store.strategy.naming.FileNamingStrategyFactory;\n+import org.springframework.data.hadoop.store.strategy.rollover.RolloverStrategy;\n+import org.springframework.data.hadoop.store.strategy.rollover.RolloverStrategyFactory;\n+import org.springframework.data.hadoop.store.support.LifecycleObjectSupport;\n+\n+/**\n+ * Base implementation of {@link PartitionDataStoreWriter}.\n+ *\n+ * @author Janne Valkealahti\n+ *\n+ * @param <T> the type of an entity to write\n+ * @param <K> the type of a partition key\n+ */\n+public abstract class AbstractPartitionDataStoreWriter<T, K> extends LifecycleObjectSupport implements PartitionDataStoreWriter<T, K> {\n+\n+\tprivate final static Log log = LogFactory.getLog(AbstractPartitionDataStoreWriter.class);\n+\n+\tprivate final Configuration configuration;\n+\n+\tprivate final Path basePath;\n+\n+\tprivate final CodecInfo codec;\n+\n+\tprivate PartitionStrategy<T, K> partitionStrategy;\n+\n+\tprivate final Map<Path, DataStoreWriter<T>> writers = new ConcurrentHashMap<Path, DataStoreWriter<T>>();\n+\n+\tprivate DataStoreWriter<T> fallbackWriter;\n+\n+\tprivate FileNamingStrategyFactory<FileNamingStrategy> fileNamingStrategyFactory;\n+\n+\tprivate RolloverStrategyFactory<RolloverStrategy> rolloverStrategyFactory;\n+\n+\tprivate long idleTimeout;\n+\n+\tprivate volatile int maxOpenAttempts = AbstractDataStreamWriter.DEFAULT_MAX_OPEN_ATTEMPTS;\n+\n+\t/** Used in-writing suffix if any */\n+\tprivate String suffix;\n+\n+\t/** Used in-writing prefix if any */\n+\tprivate String prefix;\n+\n+\t/** Flag guarding if files can be overwritten */\n+\tprivate boolean overwrite = false;\n+\n+\t/**\n+\t * Instantiates a new abstract data store partition writer.\n+\t *\n+\t * @param configuration the hadoop configuration\n+\t * @param basePath the hdfs path\n+\t * @param codec the compression codec info\n+\t * @param partitionStrategy the partition strategy\n+\t */\n+\tpublic AbstractPartitionDataStoreWriter(Configuration configuration, Path basePath, CodecInfo codec,\n+\t\t\tPartitionStrategy<T, K> partitionStrategy) {\n+\t\tsuper();\n+\t\tthis.configuration = configuration;\n+\t\tthis.basePath = basePath;\n+\t\tthis.codec = codec;\n+\t\tthis.partitionStrategy = partitionStrategy;\n+\t}\n+\n+\t@Override\n+\tpublic void write(T entity) throws IOException {\n+\t\twrite(entity, partitionStrategy.getPartitionKeyResolver().resolvePartitionKey(entity));\n+\t}\n+\n+\t@Override\n+\tpublic void flush() throws IOException {\n+\t\tfor (DataStoreWriter<?> writer : writers.values()) {\n+\t\t\ttry {\n+\t\t\t\twriter.flush();\n+\t\t\t} catch (Exception e) {\n+\t\t\t\tlog.warn(\"Writer caused exception in flush\", e);\n+\t\t\t}\n+\t\t}\n+\t\tif (fallbackWriter != null) {\n+\t\t\ttry {\n+\t\t\t\tfallbackWriter.flush();\n+\t\t\t} catch (Exception e) {\n+\t\t\t\tlog.warn(\"Writer caused exception in flush\", e);","path":"spring-hadoop-store/src/main/java/org/springframework/data/hadoop/store/output/AbstractPartitionDataStoreWriter.java","position":112,"original_position":112,"commit_id":"dd32246c9d39076c43f802127d2ce02f5fa3b5e1","original_commit_id":"dd32246c9d39076c43f802127d2ce02f5fa3b5e1","user":{"login":"markpollack","id":247466,"avatar_url":"https://avatars.githubusercontent.com/u/247466?","gravatar_id":"3c2bbcb3e173ee1b0c2a8bdb2e293633","url":"https://api.github.com/users/markpollack","html_url":"https://github.com/markpollack","followers_url":"https://api.github.com/users/markpollack/followers","following_url":"https://api.github.com/users/markpollack/following{/other_user}","gists_url":"https://api.github.com/users/markpollack/gists{/gist_id}","starred_url":"https://api.github.com/users/markpollack/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/markpollack/subscriptions","organizations_url":"https://api.github.com/users/markpollack/orgs","repos_url":"https://api.github.com/users/markpollack/repos","events_url":"https://api.github.com/users/markpollack/events{/privacy}","received_events_url":"https://api.github.com/users/markpollack/received_events","type":"User","site_admin":false},"body":"if fallbackWriter is still to be used in this class, then a different exception string as compared to the 'normal' set of DataStoreWriters should be used.","created_at":"2014-05-22T04:29:13Z","updated_at":"2014-05-22T04:29:13Z","html_url":"https://github.com/spring-projects/spring-hadoop/pull/100#discussion_r12934603","pull_request_url":"https://api.github.com/repos/spring-projects/spring-hadoop/pulls/100","_links":{"self":{"href":"https://api.github.com/repos/spring-projects/spring-hadoop/pulls/comments/12934603"},"html":{"href":"https://github.com/spring-projects/spring-hadoop/pull/100#discussion_r12934603"},"pull_request":{"href":"https://api.github.com/repos/spring-projects/spring-hadoop/pulls/100"}}}}