Commit 91369672 authored by chiaming2000's avatar chiaming2000
Browse files

1. Java API exception handling/reporting improvement - added API in 

kinetic.client.KineticException to get request/response message when a
KineticException occurred.  Applications may obtain response information
(such as status code) from the response message when a KineticException
occurred.

2. Java API throws kinetic.client.ClusterVersionFailureException when a
cluster version failure occurred.  

3. The simulator set the "current" cluster version in the response
message (Header/clusterVersion field) when a cluster version failure
occurred.  Applications may obtain the current cluster version with
ClusterVersionFailureException.getResponseMessage() when a
ClusterVersionFailureException is raised.
parent e10408e5

kinetic-client/src/main/java/com/seagate/kinetic/client/internal/ClientProxy.java

+68 −37
Original line number Diff line number Diff line
@@ -42,7 +42,7 @@ import com.seagate.kinetic.proto.Kinetic.Message.Builder; 
import com.seagate.kinetic.proto.Kinetic.Message.Header;

import com.seagate.kinetic.proto.Kinetic.Message.MessageType;

import com.seagate.kinetic.proto.Kinetic.Message.Range;

import com.seagate.kinetic.proto.Kinetic.Message.Status;





/**

 * Perform request response operation synchronously or asynchronously on behalf
@@ -241,32 +241,20 @@ public class ClientProxy { 
     *

     * @return an array of keys in db that matched the specified range.

     *

     * @throws LCException

     * @throws KineticException

     *             if any internal error occurred.

     *

     * @see KineticClient#getKeyRange(byte[], boolean, byte[], boolean, int)

     * @see KeyRange

     */

    List<ByteString> getKeyRange(KeyRange range) throws LCException {

        try {

    List<ByteString> getKeyRange(KeyRange range) throws KineticException {

        

            // perform key range op

            KineticMessage resp = doRange(range);



            // check response message type

            if (resp.getMessage().getCommand().getHeader().getMessageType() != MessageType.GETKEYRANGE_RESPONSE) {

                throw new LCException("received wrong message type.");

            }



            // return list of matched keys.

            return resp.getMessage().getCommand().getBody().getRange()

                    .getKeyList();

        } catch (LCException lce) {

            throw lce;

        } catch (Exception e) {

            throw new LCException("Message processing error: " + e

                    + e.getStackTrace()[0].toString());

        }

    }



    /**
@@ -279,15 +267,16 @@ public class ClientProxy { 
     * @throws LCException

     *             if any internal error occurred.

     */

    KineticMessage doRange(KeyRange keyRange) throws LCException {



        try {

    KineticMessage doRange(KeyRange keyRange) throws KineticException {



        //request message

        KineticMessage request = null;

        // response message

        KineticMessage respond = null;

        

        try {

            // request message

            KineticMessage request = MessageFactory

            request = MessageFactory

                    .createKineticMessageWithBuilder();

            Message.Builder msg = (Builder) request.getMessage();
@@ -310,23 +299,20 @@ public class ClientProxy { 
            // send request

            respond = request(request);

            

            // check response type

            if (respond.getMessage().getCommand().getHeader().getMessageType() != MessageType.GETKEYRANGE_RESPONSE) {

                throw new LCException("received wrong message type.");

            }



            // check status

            if (respond.getMessage().getCommand().getStatus().getCode() != Status.StatusCode.SUCCESS) {

                throw new LCException("operation failed: "

                        + respond.getMessage().getCommand().getStatus()

                        .getStatusMessage());

            }

            MessageFactory.checkReply(request, respond);



            // return response

            return respond;

        } catch (KineticException ke) {

            //re-throw ke

            throw ke;

        } catch (Exception e) {

            throw new LCException("Message processing error: " + e

                    + e.getStackTrace()[0].toString());

            //make a new kinetic exception

            KineticException ke = new KineticException (e);

            ke.setRequestMessage(request);

            ke.setResponseMessage(respond);

            //throw ke

            throw ke;

        }

    }
@@ -351,6 +337,51 @@ public class ClientProxy { 
        throw new LCException(exceptionMessage);

    }

    

    /**

     * Send a kinetic request message to drive/simulator.

     * 

     * @param krequest the request message

     * @return response the response message 

     * @throws KineticException if the command operation failed.

     * 

     * @see kinetic.client.VersionMismatchException

     * @see kinetic.client.ClusterVersionFailureException

     */

    KineticMessage request(KineticMessage krequest) throws KineticException {

        

        KineticMessage kresponse = null;

        

        try {

            kresponse = this.doRequest(krequest);

            

            //check status code

            MessageFactory.checkReply(krequest, kresponse);

            

        } catch (KineticException ke) {

            ke.setRequestMessage(krequest);

            ke.setResponseMessage(kresponse);

            throw ke;

        } catch (Exception e) {

            throwKineticException (e, krequest, kresponse);

        }

        

        return kresponse;

    }

    

    private void throwKineticException(Exception e, KineticMessage request,

            KineticMessage response) throws KineticException {



        //new instance

        KineticException ke = new KineticException (e);

        

        //set request message

        ke.setRequestMessage(request);

        //set response message

        ke.setResponseMessage(response);

        

        throw ke;

    }



    /**

     * Send the specified request message synchronously to the Kinetic service.

     *
@@ -365,7 +396,7 @@ public class ClientProxy { 
     * @see #requestAsync(com.seagate.kinetic.proto.Kinetic.Message.Builder,

     *      CallbackHandler)

     */

    KineticMessage request(KineticMessage im) throws LCException {

    KineticMessage doRequest(KineticMessage im) throws LCException {

        KineticMessage in = null;

        try {

kinetic-client/src/main/java/com/seagate/kinetic/client/internal/DefaultKineticClient.java

+106 −74
Original line number Diff line number Diff line
@@ -39,6 +39,7 @@ import com.seagate.kinetic.proto.Kinetic.Message.Builder; 
import com.seagate.kinetic.proto.Kinetic.Message.MessageType;

import com.seagate.kinetic.proto.Kinetic.Message.Synchronization;





/**

 *

 * Default KineticClient API implementation.
@@ -105,24 +106,26 @@ public class DefaultKineticClient implements AdvancedKineticClient { 


        // return entry

        Entry returnEntry = null;

        KineticMessage request = null;

        KineticMessage response = null;



        try {



            // construct put request message

            KineticMessage im = MessageFactory.createPutRequestMessage(

            request = MessageFactory.createPutRequestMessage(

                    entry, newVersion);



            // proto builder

            Message.Builder message = (Builder) im.getMessage();

            Message.Builder message = (Builder) request.getMessage();



            // set persist option

            setPersistOption(message, option);



            // send request

            KineticMessage reply = this.client.request(im);

            response = this.client.request(request);



            // check response

            MessageFactory.checkPutReply(reply, MessageType.PUT_RESPONSE);

            //MessageFactory.checkPutReply(reply, MessageType.PUT_RESPONSE);



            // construct return instance

            returnEntry = new Entry(entry.getKey(), entry.getValue(),
@@ -136,7 +139,11 @@ public class DefaultKineticClient implements AdvancedKineticClient { 
        } catch (KineticException lce) {

            throw lce;

        } catch (Exception e) {

            

            KineticException lce = new KineticException(e.getMessage(), e);

            lce.setRequestMessage(request);

            lce.setResponseMessage(response);

            

            throw lce;

        }
@@ -150,25 +157,30 @@ public class DefaultKineticClient implements AdvancedKineticClient { 
    public Entry get(byte[] key) throws KineticException {



        Entry entry = null;

        KineticMessage request = null;

        KineticMessage response = null;

        

        try {



            // create get request message

            KineticMessage request = MessageFactory.createGetRequestMessage(

            request = MessageFactory.createGetRequestMessage(

                    key, MessageType.GET);



            // send request

            KineticMessage response = this.client.request(request);

            response = this.client.request(request);



            // check response

            MessageFactory.checkGetReply(response, MessageType.GET_RESPONSE);

            //MessageFactory.checkGetReply(response, MessageType.GET_RESPONSE);



            // transform message to entry

            entry = MessageFactory.responsetoEntry(response);



        } catch (EntryNotFoundException enfe) {

            ;

        } catch (Exception e) {

            KineticException lce = new KineticException(e.getMessage(), e);

            throw lce;

            KineticException ke = new KineticException(e.getMessage(), e);

            ke.setRequestMessage(request);

            ke.setResponseMessage(response);

            throw ke;

        }



        return entry;
@@ -181,25 +193,29 @@ public class DefaultKineticClient implements AdvancedKineticClient { 
    public Entry getNext(byte[] key) throws KineticException {



        Entry entry = null;

        KineticMessage request = null;

        KineticMessage response = null;

        

        try {



            // create get request message

            KineticMessage request = MessageFactory.createGetRequestMessage(

            request = MessageFactory.createGetRequestMessage(

                    key, MessageType.GETNEXT);



            // send request

            KineticMessage response = this.client.request(request);

            response = this.client.request(request);



            // check response

            MessageFactory

            .checkGetReply(response, MessageType.GETNEXT_RESPONSE);

            //MessageFactory

            //.checkGetReply(response, MessageType.GETNEXT_RESPONSE);



            // transform message to entry

            entry = MessageFactory.responsetoEntry(response);



        } catch (EntryNotFoundException enfe) {

            ;

        } catch (Exception e) {

            KineticException lce = new KineticException(e.getMessage(), e);

            lce.setRequestMessage(request);

            lce.setResponseMessage(response);

            throw lce;

        }
@@ -213,25 +229,30 @@ public class DefaultKineticClient implements AdvancedKineticClient { 
    public Entry getPrevious(byte[] key) throws KineticException {



        Entry entry = null;

        KineticMessage request = null;

        KineticMessage response = null;



        try {



            // create get request message

            KineticMessage request = MessageFactory.createGetRequestMessage(

            request = MessageFactory.createGetRequestMessage(

                    key, MessageType.GETPREVIOUS);



            // send request

            KineticMessage response = this.client.request(request);

            response = this.client.request(request);



            // check response

            MessageFactory.checkGetReply(response,

                    MessageType.GETPREVIOUS_RESPONSE);

//            MessageFactory.checkGetReply(response,

//                    MessageType.GETPREVIOUS_RESPONSE);



            // transform message to entry

            entry = MessageFactory.responsetoEntry(response);



        } catch (EntryNotFoundException enfe) {

            ;

        } catch (Exception e) {

            KineticException lce = new KineticException(e.getMessage(), e);

            lce.setRequestMessage(request);

            lce.setResponseMessage(response);

            throw lce;

        }
@@ -264,7 +285,8 @@ public class DefaultKineticClient implements AdvancedKineticClient { 


            // convert to return type

            response = toByteArrayList(bsList);



        } catch (KineticException ke) {

            throw ke;

        } catch (Exception e) {

            KineticException lce = new KineticException(e.getMessage(), e);

            throw lce;
@@ -331,17 +353,30 @@ public class DefaultKineticClient implements AdvancedKineticClient { 
     * @throws KineticException

     */

    private boolean doDelete(KineticMessage request) throws KineticException {

        

        //deleted flag

        boolean deleted = false;

        

        KineticMessage response = null;

        

        try {

            // response message

            KineticMessage response = this.client.request(request);

            response = this.client.request(request);



            deleted = MessageFactory.checkDeleteReply(response);

            //deleted = MessageFactory.checkDeleteReply(response);

            deleted = true;



            LOG.fine("delete entry successfully ...");

        } catch (EntryNotFoundException enfe) {

            //entry not found returns false for delete command

            deleted = false;

        } catch (KineticException ke) {

           //re-throw ke

           throw ke;

        } catch (Exception e) {

            KineticException lce = new KineticException(e.getMessage(), e);

            lce.setRequestMessage(request);

            lce.setResponseMessage(response);

            throw lce;

        }
@@ -382,7 +417,7 @@ public class DefaultKineticClient implements AdvancedKineticClient { 
     * Send a request operation to the Kinetic server.

     * <p>

     *

     * @param message

     * @param request

     *            request message for a specific operation.

     *

     * @return respond message for the request operation.
@@ -391,14 +426,20 @@ public class DefaultKineticClient implements AdvancedKineticClient { 
     *             if any internal error occurred.

     */

    @Override

    public KineticMessage request(KineticMessage message)

    public KineticMessage request(KineticMessage request)

            throws KineticException {

        KineticMessage respond = null;



        try {

            respond = this.client.request(message);

            respond = this.client.request(request);

        } catch (KineticException ke) {

            throw ke;

        } catch (Exception e) {

            throw new KineticException(e.getMessage(), e);

            KineticException ke = new KineticException(e.getMessage(), e);

            ke.setRequestMessage(request);

            ke.setResponseMessage(respond);

            

            throw ke;

        }

        return respond;

    }
@@ -573,26 +614,31 @@ public class DefaultKineticClient implements AdvancedKineticClient { 
    public EntryMetadata getMetadata(byte[] key) throws KineticException {



        EntryMetadata metadata = null;

        KineticMessage request = null;

        KineticMessage response = null;

        

        try {



            // create get metadata request message

            KineticMessage request = MessageFactory.createGetMetadataMessage(

            request = MessageFactory.createGetMetadataMessage(

                    key, MessageType.GET);



            // send request

            KineticMessage response = this.client.request(request);

            response = this.client.request(request);



            // check response

            MessageFactory.checkGetReply(response, MessageType.GET_RESPONSE);

            //MessageFactory.checkGetReply(response, MessageType.GET_RESPONSE);



            // transform message to metadata

            metadata = MessageFactory.responsetoEntryMetadata(response

                    .getMessage());



        } catch (EntryNotFoundException enfe) {

            ;

        } catch (Exception e) {

            KineticException lce = new KineticException(e.getMessage(), e);

            throw lce;

            KineticException ke = new KineticException(e.getMessage(), e);

            ke.setRequestMessage(request);

            ke.setResponseMessage(response);

            throw ke;

        }



        return metadata;
@@ -628,6 +674,9 @@ public class DefaultKineticClient implements AdvancedKineticClient { 


        byte[] newVersion = null;

        

        KineticMessage request = null;

        KineticMessage response = null;



        if (entry.getEntryMetadata() != null) {

            newVersion = entry.getEntryMetadata().getVersion();

        }
@@ -635,10 +684,10 @@ public class DefaultKineticClient implements AdvancedKineticClient { 
        try {



            // construct put request message

            KineticMessage km = MessageFactory.createPutRequestMessage(

            request = MessageFactory.createPutRequestMessage(

                    entry, newVersion);



            Message.Builder message = (Builder) km.getMessage();

            Message.Builder message = (Builder) request.getMessage();



            // set force bit

            message.getCommandBuilder().getBodyBuilder().getKeyValueBuilder()
@@ -648,17 +697,21 @@ public class DefaultKineticClient implements AdvancedKineticClient { 
            setPersistOption(message, option);



            // send request

            KineticMessage reply = this.client.request(km);

            response = this.client.request(request);



            // check response

            MessageFactory.checkPutReply(reply, MessageType.PUT_RESPONSE);

            //MessageFactory.checkPutReply(reply, MessageType.PUT_RESPONSE);



            LOG.fine("put versioned successfully.");



        } catch (KineticException lce) {

            throw lce;

        } catch (Exception e) {

            

            KineticException lce = new KineticException(e.getMessage(), e);

            lce.setRequestMessage(request);

            lce.setResponseMessage(response);

            

            throw lce;

        }
@@ -769,25 +822,17 @@ public class DefaultKineticClient implements AdvancedKineticClient { 


        long time = System.currentTimeMillis();



        try {



        // create get request message

        KineticMessage request = MessageFactory.createNoOpRequestMessage();



        // send request

            KineticMessage response = this.client.request(request);

        this.client.request(request);



        // check response

            MessageFactory.checkNoOpReply(response);

        // MessageFactory.checkNoOpReply(response);



        // no op round trip time

        time = System.currentTimeMillis() - time;

        } catch (KineticException ke) {

            throw ke;

        } catch (Exception e) {

            KineticException lce = new KineticException(e.getMessage(), e);

            throw lce;

        }



        return time;

    }
@@ -797,25 +842,11 @@ public class DefaultKineticClient implements AdvancedKineticClient { 
     */

    @Override

    public void flush() throws KineticException {



        try {



            // create get request message

            KineticMessage request = MessageFactory.createFlushDataRequestMessage();



            // send request

            KineticMessage response = this.client.request(request);



            // check response

            MessageFactory.checkFushDataReply(response);



        } catch (KineticException ke) {

            throw ke;

        } catch (Exception e) {

            KineticException lce = new KineticException(e.getMessage(), e);

            throw lce;

        }



            this.client.request(request);

    }



    /**
@@ -876,7 +907,8 @@ public class DefaultKineticClient implements AdvancedKineticClient { 


            // convert to return type

            response = toByteArrayList(bsList);



        } catch (KineticException ke) {

            throw ke;

        } catch (Exception e) {

            KineticException lce = new KineticException(e.getMessage(), e);

            throw lce;

kinetic-client/src/main/java/com/seagate/kinetic/client/internal/EntryNotFoundException.java

0 → 100644
+28 −0
Original line number Diff line number Diff line 
package com.seagate.kinetic.client.internal;



import kinetic.client.KineticException;



public class EntryNotFoundException extends KineticException {



    /**

     * 

     */

    private static final long serialVersionUID = -2377497794808030692L;



    public EntryNotFoundException() {

        ;

    }



    public EntryNotFoundException(String message) {

        super(message);

    }



    public EntryNotFoundException(Throwable cause) {

        super(cause);

    }



    public EntryNotFoundException(String message, Throwable cause) {

        super(message, cause);

    }



}

kinetic-client/src/main/java/com/seagate/kinetic/client/internal/MessageFactory.java

+66 −158

File changed.

Preview size limit exceeded, changes collapsed.

kinetic-client/src/main/java/com/seagate/kinetic/client/io/MessageHandler.java

+12 −9
Original line number Diff line number Diff line
@@ -34,6 +34,7 @@ import kinetic.client.KineticException; 


import com.seagate.kinetic.client.internal.CallbackContext;

import com.seagate.kinetic.client.internal.ClientProxy;

import com.seagate.kinetic.client.internal.EntryNotFoundException;

import com.seagate.kinetic.client.internal.MessageFactory;

import com.seagate.kinetic.client.internal.async.DeleteAsyncCallbackHandler;

import com.seagate.kinetic.client.internal.async.GetAsyncCallbackHandler;
@@ -329,8 +330,7 @@ public class MessageHandler implements ClientMessageService, Runnable { 
		AsyncKineticException lce = null;



		try {

			MessageFactory.checkPutReply(context.getResponseMessage(),

					MessageType.PUT_RESPONSE);

			MessageFactory.checkReply(context.getRequestMessage(), context.getResponseMessage());

		} catch (KineticException e) {

			lce = getAsyncKineticException(context, e);

		}
@@ -344,8 +344,10 @@ public class MessageHandler implements ClientMessageService, Runnable { 
		AsyncKineticException lce = null;



		try {

			MessageFactory.checkGetReply(context.getResponseMessage(),

					messageType);

			MessageFactory.checkReply(context.getRequestMessage(), context.getResponseMessage());

		} catch (EntryNotFoundException enfe) {

		    //entry not found will return null entry to applications

		    ;

		} catch (KineticException e) {



			lce = new AsyncKineticException(lce);
@@ -363,8 +365,7 @@ public class MessageHandler implements ClientMessageService, Runnable { 
		AsyncKineticException lce = null;



		try {

			MessageFactory.checkGetKeyRangeReply(context.getResponseMessage(),

					messageType);

			MessageFactory.checkReply(context.getRequestMessage(), context.getResponseMessage());

		} catch (KineticException e) {



			lce = new AsyncKineticException(lce);
@@ -382,8 +383,10 @@ public class MessageHandler implements ClientMessageService, Runnable { 
		AsyncKineticException lce = null;



		try {

			MessageFactory.checkGetReply(context.getResponseMessage(),

					messageType);

			MessageFactory.checkReply(context.getRequestMessage(), context.getResponseMessage());

		} catch (EntryNotFoundException enfe) {

		    //entry not found will return null to applications

		    ;

		} catch (KineticException e) {



			lce = new AsyncKineticException(lce);
@@ -401,7 +404,7 @@ public class MessageHandler implements ClientMessageService, Runnable { 
		AsyncKineticException lce = null;



		try {

			MessageFactory.checkDeleteReply(context.getResponseMessage());

			MessageFactory.checkDeleteReply(context.getRequestMessage(), context.getResponseMessage());

		} catch (KineticException e) {



			lce = new AsyncKineticException(lce);