natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

rename StreamCompletionHandler to FileStatusHandler. patch by stuhood, reviwed by gdusbabek. CASSANDRA-1019 

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@948536 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Gary Dusbabek 2010-05-26 18:56:37 +00:00
parent cdf154de57
commit 15543601dc
5 changed files with 37 additions and 62 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
src/java/org/apache/cassandra/streaming/StreamCompletionHandler.java → src/java/org/apache/cassandra/streaming/FileStatusHandler.java
View File

@ -30,35 +30,42 @@ import org.apache.cassandra.db.Table;
import org.apache.cassandra.io.sstable.SSTableReader;
import org.apache.cassandra.io.sstable.SSTableWriter;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.streaming.IStreamComplete;
import org.apache.cassandra.streaming.StreamInManager;
import org.apache.cassandra.service.StorageService;
/**
* This is the callback handler that is invoked when we have
* completely received a single file from a remote host.
*
* TODO if we move this into CFS we could make addSSTables private, improving encapsulation.
* This is the callback handler that is invoked on the receiving node when a file changes status from RECEIVE to either
* FileStatus.STREAM (needs to be restreamed) or FileStatus.DELETE (successfully completed).
*/
class StreamCompletionHandler implements IStreamComplete
class FileStatusHandler
{
private static Logger logger = LoggerFactory.getLogger(StreamCompletionHandler.class);
private static Logger logger = LoggerFactory.getLogger(FileStatusHandler.class);
public void onStreamCompletion(InetAddress host, PendingFile pendingFile, FileStatus streamStatus) throws IOException
public void onStatusChange(InetAddress host, PendingFile pendingFile, FileStatus streamStatus) throws IOException
{
/* Parse the stream context and the file to the list of SSTables in the associated Column Family Store. */
if (pendingFile.getFilename().contains("-Data.db"))
if (FileStatus.StreamCompletionAction.STREAM == streamStatus.getAction())
{
// file needs to be restreamed
logger.warn("Streaming of file " + pendingFile + " from " + host + " failed, but will be retried.");
// request that the source node re-stream the file
MessagingService.instance.sendOneWay(streamStatus.makeStreamStatusMessage(), host);
return;
}
assert FileStatus.StreamCompletionAction.DELETE == streamStatus.getAction() :
"Unknown stream status: " + streamStatus.getAction();
// file was successfully streamed: if it was the last component of an sstable, assume that the rest
// have already arrived
if (pendingFile.getFilename().endsWith("-Data.db"))
{
// last component triggers add: see TODO in SSTable.getAllComponents()
String tableName = pendingFile.getDescriptor().ksname;
File file = new File( pendingFile.getFilename() );
File file = new File(pendingFile.getFilename());
String fileName = file.getName();
String [] temp = fileName.split("-");
//Open the file to see if all parts are now here
try
{
SSTableReader sstable = SSTableWriter.renameAndOpen(pendingFile.getFilename());
//TODO add a sanity check that this sstable has all its parts and is ok
Table.open(tableName).getColumnFamilyStore(temp[0]).addSSTable(sstable);
logger.info("Streaming added " + sstable.getFilename());
}
@ -68,12 +75,12 @@ class StreamCompletionHandler implements IStreamComplete
}
}
// send a StreamStatus message telling the source node it can delete this file
if (logger.isDebugEnabled())
logger.debug("Sending a streaming finished message with " + streamStatus + " to " + host);
/* Send a StreamStatus message which may require the source node to re-stream certain files. */
logger.debug("Sending a streaming finished message for " + pendingFile + " to " + host);
MessagingService.instance.sendOneWay(streamStatus.makeStreamStatusMessage(), host);
/* If we're done with everything for this host, remove from bootstrap sources */
// if all files have been received from this host, remove from bootstrap sources
if (StreamInManager.isDone(host) && StorageService.instance.isBootstrapMode())
{
StorageService.instance.removeBootstrapSource(host, pendingFile.getDescriptor().ksname);

32
src/java/org/apache/cassandra/streaming/IStreamComplete.java
View File

@ -1,32 +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.cassandra.streaming;
import java.io.IOException;
import java.net.InetAddress;
interface IStreamComplete
{
/*
* This callback if registered with the StreamContextManager is
* called when the stream from a host is completely handled.
*/
public void onStreamCompletion(InetAddress from, PendingFile pendingFile, FileStatus streamStatus) throws IOException;
}

12
src/java/org/apache/cassandra/streaming/IncomingStreamReader.java
View File

@ -69,7 +69,7 @@ public class IncomingStreamReader
{
/* Ask the source node to re-stream this file. */
streamStatus.setAction(FileStatus.StreamCompletionAction.STREAM);
handleStreamCompletion(remoteAddress.getAddress());
handleFileStatus(remoteAddress.getAddress());
/* Delete the orphaned file. */
File file = new File(pendingFile.getFilename());
file.delete();
@ -88,18 +88,18 @@ public class IncomingStreamReader
logger.debug("Removing stream context " + pendingFile);
}
fc.close();
handleStreamCompletion(remoteAddress.getAddress());
handleFileStatus(remoteAddress.getAddress());
}
}
private void handleStreamCompletion(InetAddress remoteHost) throws IOException
private void handleFileStatus(InetAddress remoteHost) throws IOException
{
/*
* Streaming is complete. If all the data that has to be received inform the sender via
* the stream completion callback so that the source may perform the requisite cleanup.
*/
IStreamComplete streamComplete = StreamInManager.getStreamCompletionHandler(remoteHost);
if (streamComplete != null)
streamComplete.onStreamCompletion(remoteHost, pendingFile, streamStatus);
FileStatusHandler handler = StreamInManager.getFileStatusHandler(remoteHost);
if (handler != null)
handler.onStatusChange(remoteHost, pendingFile, streamStatus);
}
}

12
src/java/org/apache/cassandra/streaming/StreamInManager.java
View File

@ -24,7 +24,7 @@ import java.net.InetAddress;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.Multimap;
import com.google.common.collect.Multimaps;
import org.apache.cassandra.streaming.IStreamComplete;
import org.apache.cassandra.streaming.FileStatusHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -38,7 +38,7 @@ class StreamInManager
/* Maintain in this map the status of the streams that need to be sent back to the source */
public static final Map<InetAddress, List<FileStatus>> streamStatusBag_ = new Hashtable<InetAddress, List<FileStatus>>();
/* Maintains a callback handler per endpoint to notify the app that a stream from a given endpoint has been handled */
public static final Map<InetAddress, IStreamComplete> streamNotificationHandlers_ = new HashMap<InetAddress, IStreamComplete>();
public static final Map<InetAddress, FileStatusHandler> streamNotificationHandlers_ = new HashMap<InetAddress, FileStatusHandler>();
public static final Multimap<InetAddress, PendingFile> activeStreams = Multimaps.synchronizedMultimap(HashMultimap.<InetAddress, PendingFile>create());
@ -85,7 +85,7 @@ class StreamInManager
}
/*
* This method helps determine if the StreamCompletionHandler needs
* This method helps determine if the FileStatusHandler needs
* to be invoked for the data being streamed from a source.
*/
public synchronized static boolean isDone(InetAddress key)
@ -93,17 +93,17 @@ class StreamInManager
return (ctxBag_.get(key) == null);
}
public synchronized static IStreamComplete getStreamCompletionHandler(InetAddress key)
public synchronized static FileStatusHandler getFileStatusHandler(InetAddress key)
{
return streamNotificationHandlers_.get(key);
}
public synchronized static void removeStreamCompletionHandler(InetAddress key)
public synchronized static void removeFileStatusHandler(InetAddress key)
{
streamNotificationHandlers_.remove(key);
}
public synchronized static void registerStreamCompletionHandler(InetAddress key, IStreamComplete streamComplete)
public synchronized static void registerFileStatusHandler(InetAddress key, FileStatusHandler streamComplete)
{
streamNotificationHandlers_.put(key, streamComplete);
}

2
src/java/org/apache/cassandra/streaming/StreamInitiateVerbHandler.java
View File

@ -86,7 +86,7 @@ public class StreamInitiateVerbHandler implements IVerbHandler
addStreamContext(message.getFrom(), localFile, streamStatus);
}
StreamInManager.registerStreamCompletionHandler(message.getFrom(), new StreamCompletionHandler());
StreamInManager.registerFileStatusHandler(message.getFrom(), new FileStatusHandler());
if (logger.isDebugEnabled())
logger.debug("Sending a stream initiate done message ...");
Message doneMessage = new Message(FBUtilities.getLocalAddress(), "", StorageService.Verb.STREAM_INITIATE_DONE, new byte[0] );