diff --git a/src/java/org/apache/cassandra/streaming/StreamCompletionHandler.java b/src/java/org/apache/cassandra/streaming/FileStatusHandler.java similarity index 57% rename from src/java/org/apache/cassandra/streaming/StreamCompletionHandler.java rename to src/java/org/apache/cassandra/streaming/FileStatusHandler.java index d03e036f17..f207aed32e 100644 --- a/src/java/org/apache/cassandra/streaming/StreamCompletionHandler.java +++ b/src/java/org/apache/cassandra/streaming/FileStatusHandler.java @@ -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); diff --git a/src/java/org/apache/cassandra/streaming/IStreamComplete.java b/src/java/org/apache/cassandra/streaming/IStreamComplete.java deleted file mode 100644 index 1ab771bccb..0000000000 --- a/src/java/org/apache/cassandra/streaming/IStreamComplete.java +++ /dev/null @@ -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; -} diff --git a/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java b/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java index 06f023f52b..ae69aa37ad 100644 --- a/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java +++ b/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java @@ -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); } } diff --git a/src/java/org/apache/cassandra/streaming/StreamInManager.java b/src/java/org/apache/cassandra/streaming/StreamInManager.java index c65d4fbac3..0ee8414968 100644 --- a/src/java/org/apache/cassandra/streaming/StreamInManager.java +++ b/src/java/org/apache/cassandra/streaming/StreamInManager.java @@ -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> streamStatusBag_ = new Hashtable>(); /* Maintains a callback handler per endpoint to notify the app that a stream from a given endpoint has been handled */ - public static final Map streamNotificationHandlers_ = new HashMap(); + public static final Map streamNotificationHandlers_ = new HashMap(); public static final Multimap activeStreams = Multimaps.synchronizedMultimap(HashMultimap.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); } diff --git a/src/java/org/apache/cassandra/streaming/StreamInitiateVerbHandler.java b/src/java/org/apache/cassandra/streaming/StreamInitiateVerbHandler.java index 57377bd189..e3f264051c 100644 --- a/src/java/org/apache/cassandra/streaming/StreamInitiateVerbHandler.java +++ b/src/java/org/apache/cassandra/streaming/StreamInitiateVerbHandler.java @@ -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] );