From dec1a7673472a3a208f8db74e7b59c999ceec49a Mon Sep 17 00:00:00 2001 From: Prabhu Joseph Date: Tue, 18 Jun 2019 18:03:43 +0530 Subject: [PATCH] YARN-9374. Fix HBaseTimelineWriterImpl sync writes blocks when HBase Down. --- .../storage/TestTimelineWriterHBaseDown.java | 93 ++++++++++++++++++++++ .../storage/HBaseTimelineWriterImpl.java | 18 ++++- 2 files changed, 110 insertions(+), 1 deletion(-) create mode 100644 hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice-hbase-tests/src/test/java/org/apache/hadoop/yarn/server/timelineservice/storage/TestTimelineWriterHBaseDown.java diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice-hbase-tests/src/test/java/org/apache/hadoop/yarn/server/timelineservice/storage/TestTimelineWriterHBaseDown.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice-hbase-tests/src/test/java/org/apache/hadoop/yarn/server/timelineservice/storage/TestTimelineWriterHBaseDown.java new file mode 100644 index 0000000..308b2f6 --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice-hbase-tests/src/test/java/org/apache/hadoop/yarn/server/timelineservice/storage/TestTimelineWriterHBaseDown.java @@ -0,0 +1,93 @@ +/** + * 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.hadoop.yarn.server.timelineservice.storage; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.HBaseTestingUtility; +import org.apache.hadoop.security.UserGroupInformation; +import org.apache.hadoop.test.GenericTestUtils; +import org.apache.hadoop.yarn.server.timelineservice.collector.TimelineCollectorContext; +import org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntities; +import org.apache.hadoop.yarn.api.records.timelineservice.ApplicationEntity; + +import static org.apache.hadoop.yarn.conf.YarnConfiguration.TIMELINE_SERVICE_READER_STORAGE_MONITOR_INTERVAL_MS; + +import org.junit.Test; +import org.junit.Assert; +import static org.junit.Assert.assertTrue; + +public class TestTimelineWriterHBaseDown { + + @Test(timeout=200000) + public void testTimelineWriterHBaseDown() throws Exception { + HBaseTestingUtility util = new HBaseTestingUtility(); + HBaseTimelineWriterImpl hbi = new HBaseTimelineWriterImpl(); + try { + util.startMiniCluster(); + DataGeneratorForTest.createSchema(util.getConfiguration()); + + Configuration c1 = util.getConfiguration(); + c1.setLong(TIMELINE_SERVICE_READER_STORAGE_MONITOR_INTERVAL_MS, 5000); + hbi.init(c1); + hbi.start(); + + TimelineEntities te = new TimelineEntities(); + ApplicationEntity entity = new ApplicationEntity(); + String appId = "application_1000178881110_2002"; + entity.setId(appId); + Long cTime = 1425016501000L; + entity.setCreatedTime(cTime); + te.addEntity(entity); + + hbi.write(new TimelineCollectorContext("ATS", "user1", "flow1", + "AB7822C10F1111", 1002345678919L, appId), te, + UserGroupInformation.createRemoteUser("user1")); + hbi.flush(); + assertTrue("HBase is not down", !hbi.isHBaseDown()); + + util.shutdownMiniHBaseCluster(); + GenericTestUtils.waitFor(() -> hbi.isHBaseDown(), 1000, 100000); + + boolean exceptionCaught = false; + try{ + hbi.write(new TimelineCollectorContext("ATS1", "user1", "flow2", + "AB7822C10F1111", 1002345678919L, appId), te, + UserGroupInformation.createRemoteUser("user1")); + } catch (Exception e) { + exceptionCaught = true; + } + assertTrue("HBaseStorageMonitor failed to detect HBase Down", + exceptionCaught); + + util.startMiniHBaseCluster(1, 1); + GenericTestUtils.waitFor(() -> !hbi.isHBaseDown(), 1000, 100000); + + try { + hbi.write(new TimelineCollectorContext("ATS", "user1", "flow3", + "AB7822C10F1111", 1002345678919L, appId), te, + UserGroupInformation.createRemoteUser("user1")); + } catch (Exception e) { + Assert.fail("HbaseStorageMonitor failed to detect HBase Up"); + } + } finally { + hbi.stop(); + util.shutdownMiniCluster(); + } + } + +} diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice-hbase/hadoop-yarn-server-timelineservice-hbase-client/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/HBaseTimelineWriterImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice-hbase/hadoop-yarn-server-timelineservice-hbase-client/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/HBaseTimelineWriterImpl.java index 3414a56..9f3a22f 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice-hbase/hadoop-yarn-server-timelineservice-hbase-client/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/HBaseTimelineWriterImpl.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice-hbase/hadoop-yarn-server-timelineservice-hbase-client/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/HBaseTimelineWriterImpl.java @@ -100,6 +100,7 @@ .getLogger(HBaseTimelineWriterImpl.class); private Connection conn; + private TimelineStorageMonitor storageMonitor; private TypedBufferedMutator entityTable; private TypedBufferedMutator appToFlowTable; private TypedBufferedMutator applicationTable; @@ -150,9 +151,16 @@ protected void serviceInit(Configuration conf) throws Exception { UserGroupInformation ugi = UserGroupInformation.isSecurityEnabled() ? UserGroupInformation.getLoginUser() : UserGroupInformation.getCurrentUser(); + storageMonitor = new HBaseStorageMonitor(conf); LOG.info("Initialized HBaseTimelineWriterImpl UGI to " + ugi); } + @Override + protected void serviceStart() throws Exception { + super.serviceStart(); + storageMonitor.start(); + } + /** * Stores the entire information in TimelineEntities to the timeline store. */ @@ -160,7 +168,7 @@ protected void serviceInit(Configuration conf) throws Exception { public TimelineWriteResponse write(TimelineCollectorContext context, TimelineEntities data, UserGroupInformation callerUgi) throws IOException { - + storageMonitor.checkStorageIsUp(); TimelineWriteResponse putStatus = new TimelineWriteResponse(); String clusterId = context.getClusterId(); @@ -242,6 +250,7 @@ public TimelineWriteResponse write(TimelineCollectorContext context, public TimelineWriteResponse write(TimelineCollectorContext context, TimelineDomain domain) throws IOException { + storageMonitor.checkStorageIsUp(); TimelineWriteResponse putStatus = new TimelineWriteResponse(); String clusterId = context.getClusterId(); @@ -591,6 +600,7 @@ private void store(byte[] rowKey, TimelineEntity te, @Override public TimelineWriteResponse aggregate(TimelineEntity data, TimelineAggregationTrack track) throws IOException { + storageMonitor.checkStorageIsUp(); return null; } @@ -603,6 +613,7 @@ public TimelineWriteResponse aggregate(TimelineEntity data, */ @Override public void flush() throws IOException { + storageMonitor.checkStorageIsUp(); // flush all buffered mutators entityTable.flush(); appToFlowTable.flush(); @@ -653,6 +664,11 @@ protected void serviceStop() throws Exception { LOG.info("closing the hbase Connection"); conn.close(); } + storageMonitor.stop(); super.serviceStop(); } + + public boolean isHBaseDown() { + return storageMonitor.isStorageDown(); + } } -- 2.7.4 (Apple Git-66)