fix some bugs on depanneur

Change-Id: I913728e273ab45de90a32b8e065daabf4621c9d9

diff --git a/depanneur b/depanneur
index 4a946a4fe8b706d5dc3037044925a1cba7e715c9..439330af8d66558616e46cf767847816962b5b7c 100755 (executable)
--- a/depanneur
+++ b/depanneur
@@ -2560,8 +2560,12 @@ sub readFromKafka {
     my @succeed_packages;
     my $cur_offset = getCurOffset("tizen-unified-status", $partition);
     my $packageNum = @{$packages_level{$cur_level}};
+    my %packages = map {$_ => 1} @{$packages_level{$cur_level}};
+       my $fail_num = 0;
+    my $succeed_num = 0;
+    my $curIndex = 0;
        info("current offset: $cur_offset");
-    while(@fail_packages + @succeed_packages < $packageNum ) {
+    while($fail_num + $succeed_num < $packageNum ) {
         my $messages = $consumer->fetch(
             'tizen-unified-status',
             $partition,
@@ -2571,19 +2575,25 @@ sub readFromKafka {
         foreach my $message ( @$messages ) {
             if ( $message->valid ) {
                 $cur_offset = $message->next_offset;
+                               if(!exists($packages{$message->key})) {
+                                       next;
+                               }
+                $curIndex++;
                 if($message->payload eq "failed") {
                     push(@fail_packages, $message->key);
-                                       info("package: $message->key build failed");
+                    print "[".$curIndex."/".$packageNum."] package: ".$message->key." build failed\n";
+                    $fail_num++;
                 } else {
                     push(@succeed_packages, $message->key);
-                                       info("package: $message->key build success");
+                    print "[".$curIndex."/".$packageNum."] package: ".$message->key." build success\n";
+                    $succeed_num++;
                 }
             } else {
                 info($message->error);
             }
         }
-        debug("wait for next 0.5s to check");
-        sleep(1); # sleep 0.5s for next check
+        debug("wait for next 1s to check");
+        sleep(1); # sleep 1s for next check
     }
     return @fail_packages;
 }
@@ -2872,7 +2882,7 @@ for my $pkg (`find "$srpm_repo_path" -type f -name "*.rpm" 2>/dev/null`) {
 # if build with cluster, not build local
 if ($enable_cluster == 1) {
     try {
-        $connection = Kafka::Connection->new( host => '109.123.100.144' );
+        $connection = Kafka::Connection->new( host => '109.123.100.144', timeout => 3 );
         $producer = Kafka::Producer->new( Connection => $connection );
         $consumer = Kafka::Consumer->new( Connection => $connection );
     } catch {
@@ -2894,19 +2904,30 @@ if ($enable_cluster == 1) {
     # when a level packages building return status are all succeed,
     # then go to next level packages.
        while(defined $packages_level{$cur_level}) {
-               writeToKafka($cur_level);
-        @fail_packages = readFromKafka($cur_level);
-               if( @fail_packages ) {
-                       foreach my $p ( @fail_packages ) {
-                               print "$p ";
-            }
-                       print "\n";
-                       error("these @fail_packages packages build failed");
-               }
-               else {
-                       info("@{$packages_level{$cur_level}} packages build succeed");
+       try {
+                       info("current level: $cur_level");      
+                       writeToKafka($cur_level);
+           @fail_packages = readFromKafka($cur_level);
+                       if( @fail_packages ) {
+                               foreach my $p ( @fail_packages ) {
+                                       print "$p ";
+               }
+                               print "\n";
+                               error("these @fail_packages packages build failed");
+                       }
+                       else {
+                               info("@{$packages_level{$cur_level}} packages build succeed");
+                       }
+                       $cur_level++;
+               } catch {
+                       my $error = $_;
+               if ( blessed( $error ) && $error->isa( 'Kafka::Exception' ) ) {
+                   warn 'Error: (', $error->code, ') ',  $error->message, "\n";
+                   exit;
+               } else {
+                   die $error;
+               }
                }
-               $cur_level++;
        }
        exit $ret;
 }