Duplicte Masseages In Target Queues

Hi

We have problems with duplicate messages in Service Broker queues. We have tried the "fire and forget" method.



Senario
Initiator doing SEND and END Conversation , target doing RECEIVE and End Conversation ,
Sql servar agent job runing every minute doing End conversation with cleanup in sys.conversations_endpionts queue.

We are runing 5000 - 10000 messages per minute.

When the clean up job is running we get som duplicte messages in the target queue.



Enviroment
Two separate machines runing Sql Server 2005 Standard Editon SP2


Initiator Machine


CREATE MESSAGE TYPE [TransactionStatisticsSend] AUTHORIZATION [dbo] VALIDATION = WELL_FORMED_XML

CREATE CONTRACT [TransactionStatistics] AUTHORIZATION [dbo] ([TransactionStatisticsSend] SENT BY ANY)


CREATE QUEUE [dbo].[TransactionStatisticsQueueActive] WITH STATUS = ON , RETENTION = OFF
ON [PRIMARY]

CREATE SERVICE [TransactionStatisticsServiceActive] AUTHORIZATION [dbo] ON QUEUE [dbo].[TransactionStatisticsQueueActive] ([TransactionStatistics])


CREATE ROUTE [Route::TransactionStatisticsServiceStat,0705DBB2-8CBA-43BC-A8FF-774A27F8ABC0] AUTHORIZATION [dbo] WITH SERVICE_NAME = N'TransactionStatisticsServiceStat' , ADDRESS = N'tcp://EBSDBCONFIG1A:4022'


CREATE REMOTE SERVICE BINDING [RSB::TransactionStatisticsServiceStat] AUTHORIZATION [dbo] TO SERVICE N'TransactionStatisticsServiceStat' WITH USER = [Proxy::BtsDebitServiceStat] , ANONYMOUS = OFF


CREATE ENDPOINT [EBSDBCURR1A_BROKER]
AUTHORIZATION [BTSTULLENtsappl]
STATE=STARTED
AS TCP (LISTENER_PORT = 4022, LISTENER_IP = ALL)
FOR SERVICE_BROKER (MESSAGE_FORWARDING = DISABLED
, MESSAGE_FORWARD_SIZE = 10
, AUTHENTICATION = CERTIFICATE [EBSDBCURR1A]
, ENCRYPTION = REQUIRED ALGORITHM RC4)




ALTER PROCEDURE [dbo].[Bts_SP_TransactionStatPrepare]
(@TransferID varchar(30))
AS
declare
@RowCount Int,
@ErrorSave Int,
@Msg xml,
@dialog_handle UNIQUEIDENTIFIER



Begin


BEGIN TRY

set @msg = (Select
Regtime,
SendPartShortName,
RecPartShortName,
MsgType,
From btslogactive.dbo.StatTransferlog Tl
where TransferID = @TransferID
FOR XML RAW)


BEGIN DIALOG CONVERSATION @dialog_handle
FROM SERVICE [TransactionStatisticsServiceActive]
TO SERVICE 'TransactionStatisticsServiceStat'
ON CONTRACT [TransactionStatistics]
WITH ENCRYPTION = OFF ;

SEND ON CONVERSATION @dialog_handle
MESSAGE TYPE [TransactionStatisticsSend]
(@msg) ;
END CONVERSATION @dialog_handle

END TRY
BEGIN CATCH
SELECT
ERROR_NUMBER() as ErrorNumber,
ERROR_MESSAGE() as ErrorMessage;
END CATCH

End






Target Machine

CREATE MESSAGE TYPE [TransactionStatisticsSend] AUTHORIZATION [dbo] VALIDATION = WELL_FORMED_XML


CREATE CONTRACT [TransactionStatistics] AUTHORIZATION [dbo] ([TransactionStatisticsSend] SENT BY ANY)


CREATE QUEUE [dbo].[TransactionStatisticsQueueStat] WITH STATUS = ON , RETENTION = OFF
, ACTIVATION ( STATUS = ON , PROCEDURE_NAME = [dbo].[Bts_SP_TransactionStatUpdateBroker] , MAX_QUEUE_READERS = 1 , EXECUTE AS N'dbo' ) ON [PRIMARY]


CREATE SERVICE [TransactionStatisticsServiceStat] AUTHORIZATION [dbo] ON QUEUE [dbo].[TransactionStatisticsQueueStat] ([TransactionStatistics])

CREATE ROUTE [Route::TransactionStatisticsServiceActive,D8A1A78B-CEAD-4C63-B3B3-3C986D2AB3AA] AUTHORIZATION [dbo] WITH SERVICE_NAME = N'TransactionStatisticsServiceActive' , BROKER_INSTANCE = N'D8A1A78B-CEAD-4C63-B3B3-3C986D2AB3AA' , ADDRESS = N'tcp://EBSDBCURR1A:4022'


CREATE ENDPOINT [EBSDBCONFIG1A_BROKER]
AUTHORIZATION [BTSTULLENtsappl]
STATE=STARTED
AS TCP (LISTENER_PORT = 4022, LISTENER_IP = ALL)
FOR SERVICE_BROKER (MESSAGE_FORWARDING = DISABLED
, MESSAGE_FORWARD_SIZE = 10
, AUTHENTICATION = CERTIFICATE [EBSDBCONFIG1A]
, ENCRYPTION = REQUIRED ALGORITHM RC4)





Activation procedure

ALTER PROCEDURE [dbo].[Bts_SP_TransactionStatUpdateBroker]

AS


DECLARE @MessageType sysname
DECLARE @ConversationHandle uniqueidentifier
DECLARE @MessageBody XML
DECLARE
@RegTime datetime,
@SendPartShortName varchar(30),
@RecPartShortName varchar(30),
@RC int



BEGIN TRANSACTION;
WHILE (1=1)
BEGIN
WAITFOR
(
RECEIVE TOP (1)
@MessageType = message_type_name,
@MessageBody = message_body,
@ConversationHandle = conversation_handle,
@SeqNo = message_sequence_number
FROM [TransactionStatisticsQueueStat]
), TIMEOUT 5000;

IF (@@ROWCOUNT = 0)
BEGIN
ROLLBACK TRANSACTION
RETURN
END

IF (@MessageType = 'http://schemas.microsoft.com/SQL/ServiceBroker/EndDialog')
BEGIN
END CONVERSATION @ConversationHandle
BREAK
END
ELSE IF (@MessageType = 'http://schemas.microsoft.com/SQL/ServiceBroker/Error')
BEGIN
END CONVERSATION @ConversationHandle
BREAK
END
ELSE IF (@MessageType = 'TransactionStatisticsSend')
BEGIN
Begin try
SET @RegTime = @MessageBody.value('(/row/@Regtime)[1]', 'datetime')
SET @SendPartShortName = @MessageBody.value('(/row/@SendPartShortName)[1]', 'varchar(30)')
SET @RecPartShortName = @MessageBody.value('(/row/@RecPartShortName)[1]', 'varchar(30)')
SET @MsgType = @MessageBody.value('(/row/@MsgType)[1]', 'varchar(30)')

.


End try
BEGIN CATCH
SELECT
ERROR_NUMBER() as ErrorNumber,
ERROR_MESSAGE() as ErrorMessage;
END CONVERSATION @ConversationHandle
BREAK
END CATCH

-- EXEC @RC = SP_XXXXX



END
END CONVERSATION @ConversationHandle
BREAK
END
END
COMMIT TRANSACTION
RETURN




Sql Server Agent job procedure


ALTER PROCEDURE [dbo].[Bts_SP_Del_Conversation_Endpoints]

AS
begin
DECLARE c_PurgeConversationEndpoints CURSOR FAST_FORWARD
FOR SELECT conversation_handle
FROM sys.conversation_endpoints
with (nolock)
WHERE is_system = 0
AND [State] = 'CD';

OPEN c_PurgeConversationEndpoints;
DECLARE @DialogHandle UNIQUEIDENTIFIER;
FETCH NEXT FROM c_PurgeConversationEndpoints
INTO @DialogHandle;
WHILE @@FETCH_STATUS = 0
BEGIN
END CONVERSATION @DialogHandle WITH CLEANUP;
FETCH NEXT FROM c_PurgeConversationEndpoints
INTO @DialogHandle;
END
CLOSE c_PurgeConversationEndpoints;
DEALLOCATE c_PurgeConversationEndpoints;


end